public void init ( String listenerName , Properties parameters ) throws PropertyException { try { if ( ! parameters . containsKey ( HOST_LIST ) ) throw new Exception ( "Missing bootstrap.servers property for Kafka listener" ) ; else { String [ ] hostarray = parameters . getProperty ( HOST_LIST ) . split ( "," ) ; if ( hostarray != null && hostarray . length > 0 ) hostList = Arrays . asList ( hostarray ) ; else throw new Exception ( "Missing value for bootstrap.servers property for Kafka listener" ) ; } } catch ( Exception e ) { throw new PropertyException ( e . getMessage ( ) ) ; } // STANDALONE logger = StandaloneLogger.getSingleton(); /* WITHENGINE */ logger = LoggerUtil . getStandardLogger ( ) ; logger . info ( "Starting Kafka consumer Listener name = " + listenerName ) ; kafkaListenerName = listenerName ; if ( ! parameters . containsKey ( TOPIC_LIST ) ) topics = Arrays . asList ( listenerName ) ; else { String [ ] topicList = parameters . getProperty ( TOPIC_LIST ) . split ( "," ) ; if ( topicList != null && topicList . length > 0 ) topics = Arrays . asList ( topicList ) ; else topics = Arrays . asList ( listenerName ) ; } if ( ! parameters . containsKey ( GROUP_ID ) ) { logger . warn ( "No group.id property specified for Kafka consumer " + kafkaListenerName + ", using \"" + kafkaListenerName + "\"..." ) ; parameters . put ( GROUP_ID , kafkaListenerName ) ; } if ( ! parameters . containsKey ( AUTO_COMMIT ) ) parameters . put ( AUTO_COMMIT , "true" ) ; if ( ! parameters . containsKey ( KEY_DESERIALIZER ) ) parameters . put ( KEY_DESERIALIZER , "org.apache.kafka.common.serialization.StringDeserializer" ) ; if ( ! parameters . containsKey ( VALUE_DESERIALIZER ) ) parameters . put ( VALUE_DESERIALIZER , "org.apache.kafka.common.serialization.StringDeserializer" ) ; // hostList = parameters.getProperty(HOST_LIST); auto_commit = getBooleanProperty ( parameters , AUTO_COMMIT , true ) ; poll_timeout = 1000 * getIntegerProperty ( parameters , POLL_TIMEOUT , 60 ) ; use_thread_pool = getBooleanProperty ( parameters , USE_THREAD_POOL , false ) ; xmlWrapper = parameters . getProperty ( XML_WRAPPER ) ; initParameters = parameters ; // Remove non-Kafka properties initParameters . remove ( KAFKAPOOL_CLASS_NAME ) ; initParameters . remove ( USE_THREAD_POOL ) ; initParameters . remove ( TOPIC_LIST ) ; initParameters . remove ( POLL_TIMEOUT ) ; initParameters . remove ( XML_WRAPPER ) ; }
null if processing using the same thread