question: @ SuppressWarnings ( "rawtypes" ) protected KafkaSpoutWrapper createKafkaSpout ( Map conf , TopologyContext context , SpoutOutputCollector collector , final String topic , String schemeClsName , SpoutSpec spoutSpec , Map < String , StreamDefinition > sds ) throws Exception { String kafkaBrokerZkQuorum = config . getString ( "spout.kafkaBrokerZkQuorum" ) ; BrokerHosts hosts = null ; if ( config . hasPath ( "spout.kafkaBrokerZkBasePath" ) ) { hosts = new ZkHosts ( kafkaBrokerZkQuorum , config . getString ( "spout.kafkaBrokerZkBasePath" ) ) ; } else { hosts = new ZkHosts ( kafkaBrokerZkQuorum ) ; } String transactionZkRoot = DEFAULT_STORM_KAFKA_TRANSACTION_ZK_ROOT ; if ( config . hasPath ( "spout.stormKafkaTransactionZkPath" ) ) { transactionZkRoot = config . getString ( "spout.stormKafkaTransactionZkPath" ) ; } String zkStateTransactionRelPath = DEFAULT_STORM_KAFKA_TRANSACTION_ZK_RELATIVE_PATH ; if ( config . hasPath ( "spout.stormKafkaEagleConsumer" ) ) { zkStateTransactionRelPath = config . getString ( "spout.stormKafkaEagleConsumer" ) ; } SpoutConfig spoutConfig = new SpoutConfig ( hosts , topic , transactionZkRoot , zkStateTransactionRelPath + "/" + topic + "/" + topologyId ) ; boolean stormKafkaUseSameZkQuorumWithKafkaBroker = config . getBoolean ( "spout.stormKafkaUseSameZkQuorumWithKafkaBroker" ) ; if ( stormKafkaUseSameZkQuorumWithKafkaBroker ) { ZkServerPortUtils utils = new ZkServerPortUtils ( kafkaBrokerZkQuorum ) ; spoutConfig . zkServers = utils . getZkHosts ( ) ; spoutConfig . zkPort = utils . getZkPort ( ) ; } else { ZkServerPortUtils utils = new ZkServerPortUtils ( config . getString ( "spout.stormKafkaTransactionZkQuorum" ) ) ; spoutConfig . zkServers = utils . getZkHosts ( ) ; spoutConfig . zkPort = utils . getZkPort ( ) ; } spoutConfig . stateUpdateIntervalMs = config . hasPath ( "spout.stormKafkaStateUpdateIntervalMs" ) ? config . getInt ( "spout.stormKafkaStateUpdateIntervalMs" ) : 2000 ; spoutConfig . fetchSizeBytes = config . hasPath ( "spout.stormKafkaFetchSizeBytes" ) ? config . getInt ( "spout.stormKafkaFetchSizeBytes" ) : 1048586 ; if ( config . hasPath ( "spout.stormKafkaStartOffsetTime" ) ) { spoutConfig . startOffsetTime = config . getInt ( "spout.stormKafkaStartOffsetTime" ) ; } spoutConfig . scheme = createMultiScheme ( conf , topic , schemeClsName ) ; KafkaSpoutWrapper wrapper = new KafkaSpoutWrapper ( spoutConfig , kafkaSpoutMetric ) ; SpoutOutputCollectorWrapper collectorWrapper = new SpoutOutputCollectorWrapper ( this , collector , topic , spoutSpec , numOfRouterBolts , sds , this . serializer ) ; wrapper . open ( conf , context , collectorWrapper ) ; if ( LOG . isInfoEnabled ( ) ) { LOG . info ( "create and open kafka wrapper: topic {}, scheme class{} " , topic , schemeClsName ) ; } return wrapper ; }
target:  make this method protected to make sure unit test can work well Q: Where to persist consumer state, i.e. what offset has been consumed for each topic and partition A: stormKafkaTransactionZkPath + "/" + consumerId + "/" + topic + "/" + topologyId + "/" + partitionId Note1: PartitionManager.committedPath for composing zkState path,  _spoutConfig.zkRoot + "/" + _spoutConfig.id + "/" + _partition.getId(); consumerId by default is EagleConsumer unless it is specified by "stormKafkaEagleConsumer" Note2: put topologyId as part of zkState because one topic by design can be consumed by multiple topologies so one topology needs to know processed offset for itself <p>TODO: Should avoid use Config.get in deep calling stack, should generate config bean as early as possible