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 ; }
Given a FeedParition and Plan create a Driver that will consume from the feed partition and execute the plan .