docs: public static Driver createDriver(FeedPartition feedPartition, Plan plan, MetricRegistry metricRegistry){
      populateFeedMetricInfo(plan, feedPartition, metricRegistry);
      OperatorDesc desc = plan.getRootOperator();
      Operator oper = buildOperator(desc, metricRegistry, plan.getName(), feedPartition);
      OffsetStorage offsetStorage = null;
      OffsetStorageDesc offsetDesc = plan.getOffsetStorageDesc();
      if (offsetDesc != null && feedPartition.supportsOffsetManagement()){
        offsetStorage = buildOffsetStorage(feedPartition, plan, offsetDesc);
        Offset offset = offsetStorage.findLatestPersistedOffset();
        if (offset != null){
          feedPartition.setOffset(new String(offset.serialize(), Charsets.UTF_8));
        }
      }
      CollectorProcessor cp = new CollectorProcessor();
      cp.setTupleRetry(plan.getTupleRetry());
      int offsetCommitInterval = plan.getOffsetCommitInterval();
      Driver driver = new Driver(feedPartition, oper, offsetStorage, cp, offsetCommitInterval, metricRegistry, plan.getName());
      recurseOperatorAndDriverNode(desc, driver.getDriverNode(), metricRegistry, feedPartition);
      return driver;
    }