@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;
}

@ 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 ; }
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
