{"action":"created","issue":{"url":"https://api.github.com/repos/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/20","labels_url":"https://api.github.com/repos/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/20/labels{/name}","comments_url":"https://api.github.com/repos/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/20/comments","events_url":"https://api.github.com/repos/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/20/events","html_url":"https://github.com/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/20","id":124336400,"number":20,"title":"hi","user":{"login":"sungaomeng","id":14835442,"avatar_url":"https://avatars.githubusercontent.com/u/14835442?v=3","gravatar_id":"","url":"https://api.github.com/users/sungaomeng","html_url":"https://github.com/sungaomeng","followers_url":"https://api.github.com/users/sungaomeng/followers","following_url":"https://api.github.com/users/sungaomeng/following{/other_user}","gists_url":"https://api.github.com/users/sungaomeng/gists{/gist_id}","starred_url":"https://api.github.com/users/sungaomeng/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sungaomeng/subscriptions","organizations_url":"https://api.github.com/users/sungaomeng/orgs","repos_url":"https://api.github.com/users/sungaomeng/repos","events_url":"https://api.github.com/users/sungaomeng/events{/privacy}","received_events_url":"https://api.github.com/users/sungaomeng/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":6,"created_at":"2015-12-30T11:26:55Z","updated_at":"2016-01-04T05:16:02Z","closed_at":null,"body":""},"comment":{"url":"https://api.github.com/repos/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/comments/168583646","html_url":"https://github.com/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/20#issuecomment-168583646","issue_url":"https://api.github.com/repos/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/20","id":168583646,"user":{"login":"sungaomeng","id":14835442,"avatar_url":"https://avatars.githubusercontent.com/u/14835442?v=3","gravatar_id":"","url":"https://api.github.com/users/sungaomeng","html_url":"https://github.com/sungaomeng","followers_url":"https://api.github.com/users/sungaomeng/followers","following_url":"https://api.github.com/users/sungaomeng/following{/other_user}","gists_url":"https://api.github.com/users/sungaomeng/gists{/gist_id}","starred_url":"https://api.github.com/users/sungaomeng/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/sungaomeng/subscriptions","organizations_url":"https://api.github.com/users/sungaomeng/orgs","repos_url":"https://api.github.com/users/sungaomeng/repos","events_url":"https://api.github.com/users/sungaomeng/events{/privacy}","received_events_url":"https://api.github.com/users/sungaomeng/received_events","type":"User","site_admin":false},"created_at":"2016-01-04T05:16:02Z","updated_at":"2016-01-04T05:16:02Z","body":"hi @reachkrishnaraj \r\nThank you for your help, but I'm not yet solve the problem\r\nI open Kafka consumers, on the machine can be normal consumption\r\n我的启动方法是\r\njava -Dlogback.configurationFile=/home/sungaomeng/kafka-elasticsearch-standalone-consumer/logback.xml -jar /home/sungaomeng/kafka-elasticsearch-standalone-consumer/bin/kafka-es-consumer-0.2.jar /home/sungaomeng/kafka-elasticsearch-standalone-consumer/kafkaESConsumer.properties\r\nFirst there is a message error after start, but after I trying to send a message to the kafka, but kafka - elasticsearch - standalone - consumer always return as follows\r\n“04/Jan/2016:13:04:55:835 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:04:55:839 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:04:55:841 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka”\r\n\r\n\r\n\r\n\r\n\r\nAll of the information below is my start\r\n\r\n\r\n\r\ni[root@zk2 ~]# java -Dlogback.configurationFile=/home/sungaomeng/kafka-elasticsearch-standalone-consumer/logback.xml -jar /home/sungaomeng/kafka-elasticsearch-standalone-consumer/bin/kafka-es-consumer-0.2.jar /home/sungaomeng/kafka-elasticsearch-standalone-consumer/kafkaESConsumer.properties\r\n13:02:40,410 |-INFO in ch.qos.logback.classic.LoggerContext[default] - Found resource [/home/sungaomeng/kafka-elasticsearch-standalone-consumer/logback.xml] at [file:/home/sungaomeng/kafka-elasticsearch-standalone-consumer/logback.xml]\r\n13:02:40,562 |-INFO in ch.qos.logback.classic.joran.action.ConfigurationAction - debug attribute not set\r\n13:02:40,578 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - About to instantiate appender of type [ch.qos.logback.core.ConsoleAppender]\r\n13:02:40,583 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - Naming appender as [STDOUT]\r\n13:02:40,782 |-WARN in ch.qos.logback.core.ConsoleAppender[STDOUT] - This appender no longer admits a layout as a sub-component, set an encoder instead.\r\n13:02:40,782 |-WARN in ch.qos.logback.core.ConsoleAppender[STDOUT] - To ensure compatibility, wrapping your layout in LayoutWrappingEncoder.\r\n13:02:40,782 |-WARN in ch.qos.logback.core.ConsoleAppender[STDOUT] - See also http://logback.qos.ch/codes.html#layoutInsteadOfEncoder for details\r\n13:02:40,783 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - About to instantiate appender of type [ch.qos.logback.core.rolling.RollingFileAppender]\r\n13:02:40,787 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - Naming appender as [MAIN_LOGS]\r\n13:02:40,886 |-INFO in c.q.l.core.rolling.TimeBasedRollingPolicy - Will use gz compression\r\n13:02:40,889 |-INFO in c.q.l.core.rolling.TimeBasedRollingPolicy - Will use the pattern /tmp/kafka_es_consumer-%d{yyyyMMddHH}.%i.log for the active file\r\n13:02:40,894 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1bc6a36e - The date pattern is 'yyyyMMddHH' from file name pattern '/tmp/kafka_es_consumer-%d{yyyyMMddHH}.%i.log.gz'.\r\n13:02:40,894 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1bc6a36e - Roll-over at the top of every hour.\r\n13:02:40,901 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1bc6a36e - Setting initial period to Mon Jan 04 11:44:12 CST 2016\r\n13:02:40,922 |-INFO in ch.qos.logback.core.joran.action.NestedComplexPropertyIA - Assuming default type [ch.qos.logback.classic.encoder.PatternLayoutEncoder] for [encoder] property\r\n13:02:40,950 |-INFO in ch.qos.logback.core.rolling.RollingFileAppender[MAIN_LOGS] - Active log file name: /tmp/kafka_es_consumer.log\r\n13:02:40,950 |-INFO in ch.qos.logback.core.rolling.RollingFileAppender[MAIN_LOGS] - File property is set to [/tmp/kafka_es_consumer.log]\r\n13:02:40,953 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - About to instantiate appender of type [ch.qos.logback.core.rolling.RollingFileAppender]\r\n13:02:40,953 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - Naming appender as [FAILED_EVENTS]\r\n13:02:40,965 |-INFO in c.q.l.core.rolling.TimeBasedRollingPolicy - No compression will be used\r\n13:02:40,966 |-INFO in c.q.l.core.rolling.TimeBasedRollingPolicy - Will use the pattern failed_es_events-%d{yyyy-MM-dd}.%i.log for the active file\r\n13:02:40,967 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1ff8b8f - The date pattern is 'yyyy-MM-dd' from file name pattern 'failed_es_events-%d{yyyy-MM-dd}.%i.log'.\r\n13:02:40,967 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1ff8b8f - Roll-over at midnight.\r\n13:02:40,967 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1ff8b8f - Setting initial period to Wed Dec 30 17:34:16 CST 2015\r\n13:02:40,970 |-INFO in ch.qos.logback.core.joran.action.NestedComplexPropertyIA - Assuming default type [ch.qos.logback.classic.encoder.PatternLayoutEncoder] for [encoder] property\r\n13:02:40,973 |-INFO in ch.qos.logback.core.rolling.RollingFileAppender[FAILED_EVENTS] - Active log file name: /tmp/failed_es_events.log\r\n13:02:40,973 |-INFO in ch.qos.logback.core.rolling.RollingFileAppender[FAILED_EVENTS] - File property is set to [/tmp/failed_es_events.log]\r\n13:02:40,975 |-INFO in ch.qos.logback.classic.joran.action.RootLoggerAction - Setting level of ROOT logger to INFO\r\n13:02:40,975 |-INFO in ch.qos.logback.core.joran.action.AppenderRefAction - Attaching appender named [STDOUT] to Logger[ROOT]\r\n13:02:40,976 |-INFO in ch.qos.logback.core.joran.action.AppenderRefAction - Attaching appender named [MAIN_LOGS] to Logger[ROOT]\r\n13:02:40,976 |-INFO in ch.qos.logback.classic.joran.action.ConfigurationAction - End of configuration.\r\n13:02:40,978 |-INFO in ch.qos.logback.classic.joran.JoranConfigurator@387c703b - Registering current configuration as safe fallback point\r\n\r\n04/Jan/2016:13:02:40:992 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - Initializing the Kafka Consumer config, arguments passed to the Driver: \r\n04/Jan/2016:13:02:41:505 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - /home/sungaomeng/kafka-elasticsearch-standalone-consumer/kafkaESConsumer.properties\r\n04/Jan/2016:13:02:41:508 +0800 [main] INFO  o.e.kafka.consumer.ConsumerConfig - configFile : /home/sungaomeng/kafka-elasticsearch-standalone-consumer/kafkaESConsumer.properties\r\n04/Jan/2016:13:02:41:510 +0800 [main] INFO  o.e.kafka.consumer.ConsumerConfig - Properties : {kafkaZookeeperList=localhost:2181, startOffsetFrom=RESTART, esIndex=test-1, esClusterName=ejiajie, kafkaFetchSizeMinBytes=31457280, zkConnectionTimeoutMs=15000, messageEncoding=UTF-8, partition=0, indexHandlerClass=org.elasticsearch.kafka.consumer.BasicIndexHandler, kafkaBrokersList=localhost:9092, kafkaSimpleConsumerBufferSizeBytes=31457280, zkCuratorRetryDelayMs=2000, consumerSleepBetweenFetchsMs=10, isPerfReportingEnabled=false, esHostPortList=10.44.166.81:9300, topic=topic-test, messageHandlerClass=org.elasticsearch.kafka.consumer.messageHandlers.RawMessageStringHandler, kafkaSimpleConsumerSocketTimeoutMs=10000, consumerGroupName=ESKafkaConsumerClient, startOffset=0, zkCuratorRetryTimes=3, zkSessionTimeoutMs=10000, isDryRun=false, esIndexType=kafkaESType}\r\n04/Jan/2016:13:02:41:514 +0800 [main] INFO  o.e.kafka.consumer.ConsumerConfig - Config reading complete !\r\n04/Jan/2016:13:02:41:514 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - Created the kafka consumer config, about to create a new Kafka Consumer Job\r\n04/Jan/2016:13:02:41:519 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Initializing Kafka ...\r\n04/Jan/2016:13:02:41:519 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - consumerGroupName=ESKafkaConsumerClient\r\n04/Jan/2016:13:02:41:526 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - Instantiating KafkaClient\r\n04/Jan/2016:13:02:41:526 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - ### KafkaClient Config: ###\r\n04/Jan/2016:13:02:41:526 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - kafkaZookeeperList: localhost:2181\r\n04/Jan/2016:13:02:41:527 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - kafkaBrokersList: localhost:9092\r\n04/Jan/2016:13:02:41:527 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - kafkaGroupId: ESKafkaConsumerClient\r\n04/Jan/2016:13:02:41:528 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - topic: topic-test\r\n04/Jan/2016:13:02:41:528 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - partition: 0\r\n04/Jan/2016:13:02:41:597 +0800 [main] INFO  c.n.c.f.imps.CuratorFrameworkImpl - Starting\r\nlog4j:WARN No appenders could be found for logger (org.apache.zookeeper.ZooKeeper).\r\nlog4j:WARN Please initialize the log4j system properly.\r\nlog4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.\r\n04/Jan/2016:13:02:41:698 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - Connected to Kafka Zookeeper successfully\r\n04/Jan/2016:13:02:41:698 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - Looking for Kafka leader broker ...\r\n04/Jan/2016:13:02:42:498 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - Found leader: leaderBrokerURL=zk1:9092\r\n04/Jan/2016:13:02:42:511 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - Initialized Kafka Consumer successfully\r\n04/Jan/2016:13:02:42:512 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka client created and intialized OK\r\n04/Jan/2016:13:02:42:512 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - ElasticSearch HostPortList is: 10.44.166.81:9300 \r\n04/Jan/2016:13:02:42:512 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Initializing ElasticSearch ...\r\n04/Jan/2016:13:02:42:512 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - esClusterName=ejiajie\r\n04/Jan/2016:13:02:43:735 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - adding [10.44.166.81:9300] to TransportClient ... \r\n04/Jan/2016:13:02:44:124 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - ElasticSearch Client created and intialized OK\r\n04/Jan/2016:13:02:44:125 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - MessageHandler Class given in config is org.elasticsearch.kafka.consumer.messageHandlers.RawMessageStringHandler\r\n04/Jan/2016:13:02:44:130 +0800 [main] INFO  o.e.kafka.consumer.MessageHandler - Created IndexHandler: \r\n04/Jan/2016:13:02:44:132 +0800 [main] INFO  o.e.kafka.consumer.MessageHandler - Created Message Handler\r\n04/Jan/2016:13:02:44:132 +0800 [main] INFO  o.e.k.c.m.RawMessageStringHandler - Initialized RawMessageStringHandler\r\n04/Jan/2016:13:02:44:132 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:02:44:231 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20860; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:02:44:233 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - **** Computing Kafka offset ***\r\n04/Jan/2016:13:02:44:233 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - startOffsetFrom=RESTART\r\n04/Jan/2016:13:02:44:233 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Restarting from where the Offset is left for topic [topic-test], partition [0]\r\n04/Jan/2016:13:02:44:237 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - offsetForThisRound is set to the CurrentOffset: 20860\r\n04/Jan/2016:13:02:44:239 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Resulting offsetForThisRound = 20860\r\n04/Jan/2016:13:02:44:330 +0800 [main] INFO  o.e.kafka.consumer.MessageHandler - Total # of messages in this batch: 36; # of successfully transformed and added to Index messages: 36; offsetOfNextBatch: 20896\r\n04/Jan/2016:13:02:44:331 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - posting the messages to ElasticSearch ...\r\n04/Jan/2016:13:02:45:627 +0800 [main] ERROR o.e.kafka.consumer.MessageHandler - Bulk Message Post to ElasticSearch has errors: failure in bulk execution:\r\n[0]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo47], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[1]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo48], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[2]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo49], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[3]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo4-], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[4]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo4_], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[5]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5A], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[6]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5B], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[7]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5C], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[8]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5D], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[9]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5E], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[10]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5F], message [MapperParsingException[failed to parse, document is empty]]\r\n[11]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5G], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[12]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5H], message [MapperParsingException[failed to parse, document is empty]]\r\n[13]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5I], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[14]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5J], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[15]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5K], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[16]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5L], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[17]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5M], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[18]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5N], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[19]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5O], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[root@zk2 ~]# java -Dlogback.configurationFile=/home/sungaomeng/kafka-elasticsearch-standalone-consumer/logback.xml -jar /home/sungaomeng/kafka-elasticsearch-standalone-consumer/bin/kafka-es-consumer-0.2.jar /home/sungaomeng/kafka-elasticsearch-standalone-consumer/kafkaESConsumer.properties\r\n13:02:40,410 |-INFO in ch.qos.logback.classic.LoggerContext[default] - Found resource [/home/sungaomeng/kafka-elasticsearch-standalone-consumer/logback.xml] at [file:/home/sungaomeng/kafka-elasticsearch-standalone-consumer/logback.xml]\r\n13:02:40,562 |-INFO in ch.qos.logback.classic.joran.action.ConfigurationAction - debug attribute not set\r\n13:02:40,578 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - About to instantiate appender of type [ch.qos.logback.core.ConsoleAppender]\r\n13:02:40,583 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - Naming appender as [STDOUT]\r\n13:02:40,782 |-WARN in ch.qos.logback.core.ConsoleAppender[STDOUT] - This appender no longer admits a layout as a sub-component, set an encoder instead.\r\n13:02:40,782 |-WARN in ch.qos.logback.core.ConsoleAppender[STDOUT] - To ensure compatibility, wrapping your layout in LayoutWrappingEncoder.\r\n13:02:40,782 |-WARN in ch.qos.logback.core.ConsoleAppender[STDOUT] - See also http://logback.qos.ch/codes.html#layoutInsteadOfEncoder for details\r\n13:02:40,783 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - About to instantiate appender of type [ch.qos.logback.core.rolling.RollingFileAppender]\r\n13:02:40,787 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - Naming appender as [MAIN_LOGS]\r\n13:02:40,886 |-INFO in c.q.l.core.rolling.TimeBasedRollingPolicy - Will use gz compression\r\n13:02:40,889 |-INFO in c.q.l.core.rolling.TimeBasedRollingPolicy - Will use the pattern /tmp/kafka_es_consumer-%d{yyyyMMddHH}.%i.log for the active file\r\n13:02:40,894 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1bc6a36e - The date pattern is 'yyyyMMddHH' from file name pattern '/tmp/kafka_es_consumer-%d{yyyyMMddHH}.%i.log.gz'.\r\n13:02:40,894 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1bc6a36e - Roll-over at the top of every hour.\r\n13:02:40,901 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1bc6a36e - Setting initial period to Mon Jan 04 11:44:12 CST 2016\r\n13:02:40,922 |-INFO in ch.qos.logback.core.joran.action.NestedComplexPropertyIA - Assuming default type [ch.qos.logback.classic.encoder.PatternLayoutEncoder] for [encoder] property\r\n13:02:40,950 |-INFO in ch.qos.logback.core.rolling.RollingFileAppender[MAIN_LOGS] - Active log file name: /tmp/kafka_es_consumer.log\r\n13:02:40,950 |-INFO in ch.qos.logback.core.rolling.RollingFileAppender[MAIN_LOGS] - File property is set to [/tmp/kafka_es_consumer.log]\r\n13:02:40,953 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - About to instantiate appender of type [ch.qos.logback.core.rolling.RollingFileAppender]\r\n13:02:40,953 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - Naming appender as [FAILED_EVENTS]\r\n13:02:40,965 |-INFO in c.q.l.core.rolling.TimeBasedRollingPolicy - No compression will be used\r\n13:02:40,966 |-INFO in c.q.l.core.rolling.TimeBasedRollingPolicy - Will use the pattern failed_es_events-%d{yyyy-MM-dd}.%i.log for the active file\r\n13:02:40,967 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1ff8b8f - The date pattern is 'yyyy-MM-dd' from file name pattern 'failed_es_events-%d{yyyy-MM-dd}.%i.log'.\r\n13:02:40,967 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1ff8b8f - Roll-over at midnight.\r\n13:02:40,967 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1ff8b8f - Setting initial period to Wed Dec 30 17:34:16 CST 2015\r\n13:02:40,970 |-INFO in ch.qos.logback.core.joran.action.NestedComplexPropertyIA - Assuming default type [ch.qos.logback.classic.encoder.PatternLayoutEncoder] for [encoder] property\r\n13:02:40,973 |-INFO in ch.qos.logback.core.rolling.RollingFileAppender[FAILED_EVENTS] - Active log file name: /tmp/failed_es_events.log\r\n13:02:40,973 |-INFO in ch.qos.logback.core.rolling.RollingFileAppender[FAILED_EVENTS] - File property is set to [/tmp/failed_es_events.log]\r\n13:02:40,975 |-INFO in ch.qos.logback.classic.joran.action.RootLoggerAction - Setting level of ROOT logger to INFO\r\n13:02:40,975 |-INFO in ch.qos.logback.core.joran.action.AppenderRefAction - Attaching appender named [STDOUT] to Logger[ROOT]\r\n13:02:40,976 |-INFO in ch.qos.logback.core.joran.action.AppenderRefAction - Attaching appender named [MAIN_LOGS] to Logger[ROOT]\r\n13:02:40,976 |-INFO in ch.qos.logback.classic.joran.action.ConfigurationAction - End of configuration.\r\n13:02:40,978 |-INFO in ch.qos.logback.classic.joran.JoranConfigurator@387c703b - Registering current configuration as safe fallback point\r\n\r\n04/Jan/2016:13:02:40:992 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - Initializing the Kafka Consumer config, arguments passed to the Driver: \r\n04/Jan/2016:13:02:41:505 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - /home/sungaomeng/kafka-elasticsearch-standalone-consumer/kafkaESConsumer.properties\r\n04/Jan/2016:13:02:41:508 +0800 [main] INFO  o.e.kafka.consumer.ConsumerConfig - configFile : /home/sungaomeng/kafka-elasticsearch-standalone-consumer/kafkaESConsumer.properties\r\n04/Jan/2016:13:02:41:510 +0800 [main] INFO  o.e.kafka.consumer.ConsumerConfig - Properties : {kafkaZookeeperList=localhost:2181, startOffsetFrom=RESTART, esIndex=test-1, esClusterName=ejiajie, kafkaFetchSizeMinBytes=31457280, zkConnectionTimeoutMs=15000, messageEncoding=UTF-8, partition=0, indexHandlerClass=org.elasticsearch.kafka.consumer.BasicIndexHandler, kafkaBrokersList=localhost:9092, kafkaSimpleConsumerBufferSizeBytes=31457280, zkCuratorRetryDelayMs=2000, consumerSleepBetweenFetchsMs=10, isPerfReportingEnabled=false, esHostPortList=10.44.166.81:9300, topic=topic-test, messageHandlerClass=org.elasticsearch.kafka.consumer.messageHandlers.RawMessageStringHandler, kafkaSimpleConsumerSocketTimeoutMs=10000, consumerGroupName=ESKafkaConsumerClient, startOffset=0, zkCuratorRetryTimes=3, zkSessionTimeoutMs=10000, isDryRun=false, esIndexType=kafkaESType}\r\n04/Jan/2016:13:02:41:514 +0800 [main] INFO  o.e.kafka.consumer.ConsumerConfig - Config reading complete !\r\n04/Jan/2016:13:02:41:514 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - Created the kafka consumer config, about to create a new Kafka Consumer Job\r\n04/Jan/2016:13:02:41:519 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Initializing Kafka ...\r\n04/Jan/2016:13:02:41:519 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - consumerGroupName=ESKafkaConsumerClient\r\n04/Jan/2016:13:02:41:526 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - Instantiating KafkaClient\r\n04/Jan/2016:13:02:41:526 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - ### KafkaClient Config: ###\r\n04/Jan/2016:13:02:41:526 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - kafkaZookeeperList: localhost:2181\r\n04/Jan/2016:13:02:41:527 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - kafkaBrokersList: localhost:9092\r\n04/Jan/2016:13:02:41:527 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - kafkaGroupId: ESKafkaConsumerClient\r\n04/Jan/2016:13:02:41:528 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - topic: topic-test\r\n04/Jan/2016:13:02:41:528 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - partition: 0\r\n04/Jan/2016:13:02:41:597 +0800 [main] INFO  c.n.c.f.imps.CuratorFrameworkImpl - Starting\r\nlog4j:WARN No appenders could be found for logger (org.apache.zookeeper.ZooKeeper).\r\nlog4j:WARN Please initialize the log4j system properly.\r\nlog4j:WARN See http://logging.apache.org/log4j/1.2/faq.html#noconfig for more info.\r\n04/Jan/2016:13:02:41:698 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - Connected to Kafka Zookeeper successfully\r\n04/Jan/2016:13:02:41:698 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - Looking for Kafka leader broker ...\r\n04/Jan/2016:13:02:42:498 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - Found leader: leaderBrokerURL=zk1:9092\r\n04/Jan/2016:13:02:42:511 +0800 [main] INFO  o.e.kafka.consumer.KafkaClient - Initialized Kafka Consumer successfully\r\n04/Jan/2016:13:02:42:512 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka client created and intialized OK\r\n04/Jan/2016:13:02:42:512 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - ElasticSearch HostPortList is: 10.44.166.81:9300 \r\n04/Jan/2016:13:02:42:512 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Initializing ElasticSearch ...\r\n04/Jan/2016:13:02:42:512 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - esClusterName=ejiajie\r\n04/Jan/2016:13:02:43:735 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - adding [10.44.166.81:9300] to TransportClient ... \r\n04/Jan/2016:13:02:44:124 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - ElasticSearch Client created and intialized OK\r\n04/Jan/2016:13:02:44:125 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - MessageHandler Class given in config is org.elasticsearch.kafka.consumer.messageHandlers.RawMessageStringHandler\r\n04/Jan/2016:13:02:44:130 +0800 [main] INFO  o.e.kafka.consumer.MessageHandler - Created IndexHandler: \r\n04/Jan/2016:13:02:44:132 +0800 [main] INFO  o.e.kafka.consumer.MessageHandler - Created Message Handler\r\n04/Jan/2016:13:02:44:132 +0800 [main] INFO  o.e.k.c.m.RawMessageStringHandler - Initialized RawMessageStringHandler\r\n04/Jan/2016:13:02:44:132 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:02:44:231 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20860; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:02:44:233 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - **** Computing Kafka offset ***\r\n04/Jan/2016:13:02:44:233 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - startOffsetFrom=RESTART\r\n04/Jan/2016:13:02:44:233 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Restarting from where the Offset is left for topic [topic-test], partition [0]\r\n04/Jan/2016:13:02:44:237 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - offsetForThisRound is set to the CurrentOffset: 20860\r\n04/Jan/2016:13:02:44:239 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Resulting offsetForThisRound = 20860\r\n04/Jan/2016:13:02:44:330 +0800 [main] INFO  o.e.kafka.consumer.MessageHandler - Total # of messages in this batch: 36; # of successfully transformed and added to Index messages: 36; offsetOfNextBatch: 20896\r\n04/Jan/2016:13:02:44:331 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - posting the messages to ElasticSearch ...\r\n04/Jan/2016:13:02:45:627 +0800 [main] ERROR o.e.kafka.consumer.MessageHandler - Bulk Message Post to ElasticSearch has errors: failure in bulk execution:\r\n[0]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo47], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[1]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo48], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[2]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo49], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[3]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo4-], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[4]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo4_], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[5]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5A], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[6]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5B], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[7]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5C], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[8]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5D], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[9]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5E], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[10]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5F], message [MapperParsingException[failed to parse, document is empty]]\r\n[11]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5G], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[12]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5H], message [MapperParsingException[failed to parse, document is empty]]\r\n[13]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5I], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[14]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5J], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[15]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5K], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[16]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5L], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[17]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5M], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[18]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5N], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n[19]: index [test-1], type [kafkaESType], id [AVILBB55aguIG2YLgo5O], message [MapperParsingException[failed to parse]; nested: ElasticsearchParseException[Failed to derive xcontent]; ]\r\n04/Jan/2016:13:02:45:649 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Commiting offset: 20896 \r\n04/Jan/2016:13:02:45:679 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - *** Finished current round of ConsumerJob, processed messages with offsets [20860-20896] ****\r\n04/Jan/2016:13:02:55:680 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:02:55:685 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:02:55:687 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n04/Jan/2016:13:03:05:689 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:03:05:696 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:03:05:697 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n04/Jan/2016:13:03:15:699 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:03:15:704 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:03:15:705 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n04/Jan/2016:13:03:25:710 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:03:25:714 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:03:25:716 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n04/Jan/2016:13:03:35:717 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:03:35:722 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:03:35:725 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n04/Jan/2016:13:03:45:727 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:03:45:733 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:03:45:735 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n04/Jan/2016:13:03:55:763 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:03:55:769 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:03:55:771 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n04/Jan/2016:13:04:05:772 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:04:05:778 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:04:05:780 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n04/Jan/2016:13:04:15:781 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:04:15:798 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:04:15:801 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n04/Jan/2016:13:04:25:803 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:04:25:808 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:04:25:809 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n04/Jan/2016:13:04:35:811 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:04:35:817 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:04:35:819 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n04/Jan/2016:13:04:45:825 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:04:45:830 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:04:45:832 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n04/Jan/2016:13:04:55:835 +0800 [main] INFO  o.e.k.c.daemon.KafkaConsumerMain - ******* Starting a new round of kafka consumer job\r\n04/Jan/2016:13:04:55:839 +0800 [main] INFO  o.e.kafka.consumer.ConsumerJob - Kafka offsets: currentOffset=20896; earliestOffset=20342; latestOffset=20896\r\n04/Jan/2016:13:04:55:841 +0800 [main] WARN  o.e.kafka.consumer.ConsumerJob - No events were read from Kafka - finishing this round of reads from Kafka\r\n^CRunning Shutdown Hook .... \r\n04/Jan/2016:13:04:56:400 +0800 [Thread-0] INFO  o.e.k.c.daemon.KafkaConsumerMain - Received the stop signal, trying to stop the Consumer job\r\n04/Jan/2016:13:04:56:445 +0800 [Thread-0] INFO  o.e.k.c.daemon.KafkaConsumerMain - Completed waiting for inprogess Consumer Job to finish - stopping the job\r\n04/Jan/2016:13:04:56:447 +0800 [Thread-0] INFO  o.e.kafka.consumer.ConsumerJob - About to close the Kafka & ElasticSearch Client\r\n04/Jan/2016:13:04:56:460 +0800 [Thread-0] INFO  o.e.kafka.consumer.KafkaClient - Curator/Zookeeper connection closed\r\n04/Jan/2016:13:04:56:525 +0800 [Thread-0] INFO  o.e.kafka.consumer.ConsumerJob - Closed Kafka & ElasticSearch Client\r\n04/Jan/2016:13:04:56:525 +0800 [Thread-0] INFO  o.e.k.c.daemon.KafkaConsumerMain - Stopped the Consumer Job\r\n[root@zk2 ~]# \r\n\r\n\r\n:9200/_plugin/head/  Display is normal, but the index is 575 b\r\n\r\nNo matter how to thank you    Krishna Raj\r\n"}}