{"action":"created","issue":{"url":"https://api.github.com/repos/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/23","labels_url":"https://api.github.com/repos/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/23/labels{/name}","comments_url":"https://api.github.com/repos/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/23/comments","events_url":"https://api.github.com/repos/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/23/events","html_url":"https://github.com/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/23","id":126311348,"number":23,"title":"Failed to post messages to ElasticSearch: None of the configured nodes are available: []","user":{"login":"3quanfeng","id":6904612,"avatar_url":"https://avatars.githubusercontent.com/u/6904612?v=3","gravatar_id":"","url":"https://api.github.com/users/3quanfeng","html_url":"https://github.com/3quanfeng","followers_url":"https://api.github.com/users/3quanfeng/followers","following_url":"https://api.github.com/users/3quanfeng/following{/other_user}","gists_url":"https://api.github.com/users/3quanfeng/gists{/gist_id}","starred_url":"https://api.github.com/users/3quanfeng/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/3quanfeng/subscriptions","organizations_url":"https://api.github.com/users/3quanfeng/orgs","repos_url":"https://api.github.com/users/3quanfeng/repos","events_url":"https://api.github.com/users/3quanfeng/events{/privacy}","received_events_url":"https://api.github.com/users/3quanfeng/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"milestone":null,"comments":2,"created_at":"2016-01-13T00:31:07Z","updated_at":"2016-01-13T01:06:59Z","closed_at":null,"body":"Hello guys, thanks for the tool you guys created, it's really awesome.\r\nI just downloaded it yesterday and started to play with it....\r\nI got some issue when I run this tool to read event(json) from broker and write them to ES.\r\nI set up all the things in the same box(an ec2 instance) including kafka(2.10.0-0.8.2.1), kafka-elasticsearch-standalone-consumer(branch 2.0) and Elasticsearch(2.1.1)\r\n\r\n\r\nCreate a \"test\" topic, only 1 partition\r\nbin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test\r\n\r\nSend message:\r\n\r\n[ec2-user@xxxx kafka_2.10-0.8.2.1]$ bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test\r\n[2016-01-12 23:46:44,492] WARN Property topic is not valid (kafka.utils.VerifiableProperties)\r\n{\"index\":{\"_id\":\"6\"}}\r\n{\"account_number\":6,\"balance\":5686,\"firstname\":\"Hattie\",\"lastname\":\"Bond\",\"age\":36,\"gender\":\"M\",\"address\":\"671 Bristol Street\",\"employer\":\"Netagy\",\"email\":\"\r\nhattiebond@netagy.com\",\"city\":\"Dante\",\"state\":\"TN\"}\r\n\r\nAnd then run kafka-elasticsearch-standalone-consumer(config is provided below)\r\nHERE IS THE LOG:\r\n23:56:43,607 |-INFO in ch.qos.logback.classic.LoggerContext[default] - Found resource [/home/ec2-user/apps/kafka/logback.xml] at [file:/home/ec2-user/apps/kafka/logback.xml]\r\n23:56:43,708 |-INFO in ch.qos.logback.classic.joran.action.ConfigurationAction - debug attribute not set\r\n23:56:43,729 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - About to instantiate appender of type [ch.qos.logback.core.ConsoleAppender]\r\n23:56:43,734 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - Naming appender as [STDOUT]\r\n23:56:43,863 |-WARN in ch.qos.logback.core.ConsoleAppender[STDOUT] - This appender no longer admits a layout as a sub-component, set an encoder instead.\r\n23:56:43,863 |-WARN in ch.qos.logback.core.ConsoleAppender[STDOUT] - To ensure compatibility, wrapping your layout in LayoutWrappingEncoder.\r\n23:56:43,863 |-WARN in ch.qos.logback.core.ConsoleAppender[STDOUT] - See also http://logback.qos.ch/codes.html#layoutInsteadOfEncoder for details\r\n23:56:43,863 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - About to instantiate appender of type [ch.qos.logback.core.rolling.RollingFileAppender]\r\n23:56:43,867 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - Naming appender as [MAIN_LOGS]\r\n23:56:43,914 |-INFO in c.q.l.core.rolling.TimeBasedRollingPolicy - Will use gz compression\r\n23:56:43,915 |-INFO in c.q.l.core.rolling.TimeBasedRollingPolicy - Will use the pattern /tmp/kafka_es_indexer-%d{yyyy-MM-dd}.%i.log for the active file\r\n23:56:43,918 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@63e2203c - The date pattern is 'yyyy-MM-dd' from file name pattern '/tmp/kafka_es_indexer-%d{yyyy-MM-dd}.%i.log.gz'.\r\n23:56:43,919 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@63e2203c - Roll-over at midnight.\r\n23:56:43,923 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@63e2203c - Setting initial period to Tue Jan 12 23:56:25 UTC 2016\r\n23:56:43,930 |-INFO in ch.qos.logback.core.joran.action.NestedComplexPropertyIA - Assuming default type [ch.qos.logback.classic.encoder.PatternLayoutEncoder] for [encoder] property\r\n23:56:43,945 |-INFO in ch.qos.logback.core.rolling.RollingFileAppender[MAIN_LOGS] - Active log file name: /tmp/kafka_es_indexer.log\r\n23:56:43,945 |-INFO in ch.qos.logback.core.rolling.RollingFileAppender[MAIN_LOGS] - File property is set to [/tmp/kafka_es_indexer.log]\r\n23:56:43,946 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - About to instantiate appender of type [ch.qos.logback.core.rolling.RollingFileAppender]\r\n23:56:43,946 |-INFO in ch.qos.logback.core.joran.action.AppenderAction - Naming appender as [FAILED_EVENTS]\r\n23:56:43,949 |-INFO in c.q.l.core.rolling.TimeBasedRollingPolicy - No compression will be used\r\n23:56:43,949 |-INFO in c.q.l.core.rolling.TimeBasedRollingPolicy - Will use the pattern /tmp/failed_es_events-%d{yyyy-MM-dd}.%i.log for the active file\r\n23:56:43,951 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1efed156 - The date pattern is 'yyyy-MM-dd' from file name pattern '/tmp/failed_es_events-%d{yyyy-MM-dd}.%i.log'.\r\n23:56:43,951 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1efed156 - Roll-over at midnight.\r\n23:56:43,951 |-INFO in ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP@1efed156 - Setting initial period to Tue Jan 12 22:19:15 UTC 2016\r\n23:56:43,956 |-INFO in ch.qos.logback.core.joran.action.NestedComplexPropertyIA - Assuming default type [ch.qos.logback.classic.encoder.PatternLayoutEncoder] for [encoder] property\r\n23:56:43,957 |-INFO in ch.qos.logback.core.rolling.RollingFileAppender[FAILED_EVENTS] - Active log file name: /tmp/failed_es_events.log\r\n23:56:43,957 |-INFO in ch.qos.logback.core.rolling.RollingFileAppender[FAILED_EVENTS] - File property is set to [/tmp/failed_es_events.log]\r\n23:56:43,958 |-INFO in ch.qos.logback.classic.joran.action.LoggerAction - Setting level of logger [org.elasticsearch.kafka.indexer.FailedEventsLogger] to ERROR\r\n23:56:43,958 |-INFO in ch.qos.logback.core.joran.action.AppenderRefAction - Attaching appender named [FAILED_EVENTS] to Logger[org.elasticsearch.kafka.indexer.FailedEventsLogger]\r\n23:56:43,959 |-INFO in ch.qos.logback.classic.joran.action.LoggerAction - Setting level of logger [org.elasticsearch.kafka] to INFO\r\n23:56:43,959 |-INFO in ch.qos.logback.core.joran.action.AppenderRefAction - Attaching appender named [MAIN_LOGS] to Logger[org.elasticsearch.kafka]\r\n23:56:44,334 |-INFO in ch.qos.logback.core.joran.action.AppenderRefAction - Attaching appender named [STDOUT] to Logger[org.elasticsearch.kafka]\r\n23:56:44,334 |-INFO in ch.qos.logback.classic.joran.action.ConfigurationAction - End of configuration.\r\n23:56:44,336 |-INFO in ch.qos.logback.classic.joran.JoranConfigurator@6737fd8f - Registering current configuration as safe fallback point\r\n\r\n12/Jan/2016:23:56:44:343 +0000 [main] INFO  o.e.kafka.indexer.KafkaIndexerDriver - Initializing Kafka ES Indexer, arguments passed to the Driver: \r\n12/Jan/2016:23:56:44:346 +0000 [main] INFO  o.e.kafka.indexer.KafkaIndexerDriver - /home/ec2-user/apps/kafka/kafka-es-indexer.properties\r\n12/Jan/2016:23:56:44:347 +0000 [main] INFO  o.e.kafka.indexer.ConsumerConfig - configFile : /home/ec2-user/apps/kafka/kafka-es-indexer.properties\r\n12/Jan/2016:23:56:44:348 +0000 [main] INFO  o.e.kafka.indexer.ConsumerConfig - Properties : {esIndexingRetrySleepTimeMs=10000, kafkaZookeeperList=localhost:2181, appStopTimeoutSeconds=10, esIndex=kafkaESIndex, startOffsetFrom=RESTART, esClusterName=sca, kafkaFetchSizeMinBytes=31457280, zkConnectionTimeoutMs=15000, messageEncoding=UTF-8, indexHandlerClass=org.elasticsearch.kafka.indexer.BasicIndexHandler, kafkaBrokersList=localhost:9092, kafkaSimpleConsumerBufferSizeBytes=31457280, zkCuratorRetryDelayMs=2000, consumerSleepBetweenFetchsMs=10, kafkaReinitSleepTimeMs=10000, isPerfReportingEnabled=false, esHostPortList=localhost:9300, topic=test, messageHandlerClass=org.elasticsearch.kafka.indexer.messageHandlers.RawMessageStringHandler, lastPartition=0, kafkaSimpleConsumerSocketTimeoutMs=10000, numberOfReinitTries=2, consumerGroupName=kafka_es_indexer, startOffset=0, firstPartition=0, numberOfEsIndexingRetryAttempts=2, zkCuratorRetryTimes=3, zkSessionTimeoutMs=10000, isDryRun=false, esIndexType=kafkaESType}\r\n12/Jan/2016:23:56:44:349 +0000 [main] INFO  o.e.kafka.indexer.ConsumerConfig - Config reading complete !\r\n12/Jan/2016:23:56:44:349 +0000 [main] INFO  o.e.kafka.indexer.KafkaIndexerDriver - Created kafka consumer config OK\r\n12/Jan/2016:23:56:44:350 +0000 [main] INFO  o.e.k.indexer.jobs.IndexerJobManager - ConsumerJobManager is starting, servicing partitions: [0-0]\r\n12/Jan/2016:23:56:44:351 +0000 [main] INFO  o.e.kafka.indexer.KafkaIndexerDriver - Registering KafkfaEsIndexerStatus MBean: \r\n12/Jan/2016:23:56:44:445 +0000 [main] INFO  o.e.k.indexer.jobs.IndexerJobManager - Creating IndexerJob for partition=0\r\n12/Jan/2016:23:56:44:450 +0000 [main] INFO  o.e.kafka.indexer.jobs.IndexerJob - Initializing ElasticSearch... hostPortList=localhost:9300, esClusterName=sca for partition 0\r\nlog4j:WARN No appenders could be found for logger (org.elasticsearch.plugins).\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\n12/Jan/2016:23:56:45:294 +0000 [main] INFO  o.e.kafka.indexer.jobs.IndexerJob - adding [localhost:9300] to TransportClient for partition 0... \r\n12/Jan/2016:23:56:45:474 +0000 [main] INFO  o.e.kafka.indexer.jobs.IndexerJob - ElasticSearch Client created and intialized OK for partition 0\r\n12/Jan/2016:23:56:45:474 +0000 [main] INFO  o.e.kafka.indexer.jobs.IndexerJob - Initializing Kafka for partition 0...\r\n12/Jan/2016:23:56:45:474 +0000 [main] INFO  o.e.kafka.indexer.jobs.IndexerJob - kafkaClientId=kafka_es_indexer_0 for partition 0\r\n12/Jan/2016:23:56:45:476 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - Instantiating KafkaClient\r\n12/Jan/2016:23:56:45:476 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - ### KafkaClient Config: ###\r\n12/Jan/2016:23:56:45:476 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - kafkaZookeeperList: localhost:2181\r\n12/Jan/2016:23:56:45:476 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - kafkaBrokersList: localhost:9092\r\n12/Jan/2016:23:56:45:477 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - kafkaClientId: kafka_es_indexer_0\r\n12/Jan/2016:23:56:45:477 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - topic: test\r\n12/Jan/2016:23:56:45:477 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - partition: 0\r\n12/Jan/2016:23:56:45:515 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - Connected to Kafka Zookeeper successfully\r\n12/Jan/2016:23:56:45:515 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - Looking for Kafka leader broker for partition 0...\r\n12/Jan/2016:23:56:45:515 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - Looking for leader for partition 0 using Kafka Broker=localhost:9092, topic=test\r\n12/Jan/2016:23:56:45:985 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - Found leader for partition 0 using Kafka Broker=localhost:9092, topic=test; leader broker URL: 0.0.0.0:9092\r\n12/Jan/2016:23:56:45:987 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - Found leader: leaderBrokerURL=0.0.0.0:9092\r\n12/Jan/2016:23:56:45:990 +0000 [main] INFO  o.e.kafka.indexer.KafkaClient - Initialized Kafka Consumer successfully for partition 0\r\n12/Jan/2016:23:56:45:990 +0000 [main] INFO  o.e.kafka.indexer.jobs.IndexerJob - Kafka client created and intialized OK for partition 0\r\n12/Jan/2016:23:56:45:990 +0000 [main] INFO  o.e.kafka.indexer.jobs.IndexerJob - MessageHandler Class given in config is org.elasticsearch.kafka.indexer.messageHandlers.RawMessageStringHandler for partition 0\r\n12/Jan/2016:23:56:45:993 +0000 [main] INFO  o.e.kafka.indexer.MessageHandler - Created IndexHandler: \r\n12/Jan/2016:23:56:45:994 +0000 [main] INFO  o.e.kafka.indexer.MessageHandler - Created Message Handler\r\n12/Jan/2016:23:56:45:994 +0000 [main] INFO  o.e.k.i.m.RawMessageStringHandler - Initialized RawMessageStringHandler\r\n12/Jan/2016:23:56:45:995 +0000 [kafka-indexer-consumer-thread-0] INFO  o.e.kafka.indexer.jobs.IndexerJob - ******* Starting a new batch of events from Kafka for partition 0 ...\r\n12/Jan/2016:23:56:45:995 +0000 [kafka-indexer-consumer-thread-0] INFO  o.e.kafka.indexer.jobs.IndexerJob - **** Computing Kafka offset *** for partition 0\r\n12/Jan/2016:23:56:45:996 +0000 [kafka-indexer-consumer-thread-0] INFO  o.e.kafka.indexer.jobs.IndexerJob - startOffsetFrom=RESTART for partition 0\r\n12/Jan/2016:23:56:45:996 +0000 [kafka-indexer-consumer-thread-0] INFO  o.e.kafka.indexer.jobs.IndexerJob - Restarting from where the Offset is left for topic test, for partition 0\r\n12/Jan/2016:23:56:46:012 +0000 [kafka-indexer-consumer-thread-0] INFO  o.e.kafka.indexer.jobs.IndexerJob - offsetForThisRound is set to the CurrentOffset: 0 for partition 0\r\n12/Jan/2016:23:56:46:038 +0000 [kafka-indexer-consumer-thread-0] INFO  o.e.kafka.indexer.jobs.IndexerJob - Resulting offsetForThisRound = 0 for partition 0\r\n12/Jan/2016:23:56:46:071 +0000 [kafka-indexer-consumer-thread-0] INFO  o.e.kafka.indexer.MessageHandler - Total # of messages in this batch: 6; # of successfully transformed and added to Index messages: 6; offsetOfNextBatch: 6\r\n12/Jan/2016:23:56:46:071 +0000 [kafka-indexer-consumer-thread-0] INFO  o.e.kafka.indexer.jobs.IndexerJob - posting the messages to ElasticSearch for partition 0...\r\n12/Jan/2016:23:56:46:077 +0000 [kafka-indexer-consumer-thread-0] ERROR o.e.kafka.indexer.MessageHandler - Failed to post messages to ElasticSearch: None of the configured nodes are available: []\r\norg.elasticsearch.client.transport.NoNodeAvailableException: None of the configured nodes are available: []\r\n\r\n\r\nMY CONFIG:\r\nkafkaZookeeperList=localhost:2181\r\nzkSessionTimeoutMs=10000\r\nzkConnectionTimeoutMs=15000\r\nzkCuratorRetryTimes=3\r\nzkCuratorRetryDelayMs=2000\r\nkafkaBrokersList=localhost:9092\r\nconsumerGroupName=kafka_es_indexer\r\ntopic=test\r\nfirstPartition=0\r\nlastPartition=0\r\nstartOffsetFrom=RESTART\r\nstartOffset=0\r\nkafkaFetchSizeMinBytes=31457280\r\nkafkaSimpleConsumerBufferSizeBytes=31457280\r\nkafkaSimpleConsumerSocketTimeoutMs=10000\r\nappStopTimeoutSeconds=10\r\nnumberOfReinitTries=2\r\nkafkaReinitSleepTimeMs=10000\r\nesHostPortList=localhost:9300\r\nesClusterName=sca\r\nesIndex=kafkaESIndex\r\nesIndexType=kafkaESType\r\nmessageHandlerClass=org.elasticsearch.kafka.indexer.messageHandlers.RawMessageStringHandler\r\nindexHandlerClass=org.elasticsearch.kafka.indexer.BasicIndexHandler\r\nmessageEncoding=UTF-8\r\nisDryRun=false\r\nconsumerSleepBetweenFetchsMs=10\r\nnumberOfEsIndexingRetryAttempts=2\r\nesIndexingRetrySleepTimeMs=10000\r\nisPerfReportingEnabled=false\r\n\r\n\r\nTROUBLESHOOTING:\r\n\r\nI acutually verified that I can run ElasticSearch API manually to do the bulk post with different index and it just works perfectly:\r\n\r\ncurl -XPOST 'localhost:9200/customer/external/_bulk?pretty' -d '\r\n{\"index\":{\"_id\":\"1\"}}\r\n{\"name\": \"John Doe\" }\r\n{\"index\":{\"_id\":\"2\"}}\r\n{\"name\": \"Jane Doe\" }\r\n'\r\n\r\nHere is the code I can find where throw the exception:\r\n\r\npublic boolean postToElasticSearch() throws Exception {\r\n\t\tBulkResponse bulkResponse = null;\r\n\t\tBulkItemResponse bulkItemResp = null;\r\n\t\t//Nothing/NoMessages to post to ElasticSearch\r\n\t\tif(bulkRequestBuilder.numberOfActions() <= 0){\r\n\t\t\tlogger.warn(\"No messages to post to ElasticSearch - returning\");\r\n\t\t\treturn true;\r\n\t\t}\t\t\r\n\t\ttry{\r\n\t\t\tbulkResponse = bulkRequestBuilder.execute().actionGet();\r\n\t\t}\r\n\t\tcatch(ElasticsearchException e){\r\n\t\t\tlogger.error(\"Failed to post messages to ElasticSearch: \" + e.getMessage(), e);\r\n\t\t\tthrow e;\r\n\t\t}\r\n                ....\r\n\r\n\r\nthe elasticsearch port is 9300 which is used in the config...\r\nI tried 9200 as well, and it didn't work either....\r\n\r\nPlease help look into the issue, I don't think it would be a connection issue to elasticsearch since they are in the same box.\r\n\r\nThanks in advance!\r\n"},"comment":{"url":"https://api.github.com/repos/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/comments/171119042","html_url":"https://github.com/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/23#issuecomment-171119042","issue_url":"https://api.github.com/repos/reachkrishnaraj/kafka-elasticsearch-standalone-consumer/issues/23","id":171119042,"user":{"login":"3quanfeng","id":6904612,"avatar_url":"https://avatars.githubusercontent.com/u/6904612?v=3","gravatar_id":"","url":"https://api.github.com/users/3quanfeng","html_url":"https://github.com/3quanfeng","followers_url":"https://api.github.com/users/3quanfeng/followers","following_url":"https://api.github.com/users/3quanfeng/following{/other_user}","gists_url":"https://api.github.com/users/3quanfeng/gists{/gist_id}","starred_url":"https://api.github.com/users/3quanfeng/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/3quanfeng/subscriptions","organizations_url":"https://api.github.com/users/3quanfeng/orgs","repos_url":"https://api.github.com/users/3quanfeng/repos","events_url":"https://api.github.com/users/3quanfeng/events{/privacy}","received_events_url":"https://api.github.com/users/3quanfeng/received_events","type":"User","site_admin":false},"created_at":"2016-01-13T01:06:59Z","updated_at":"2016-01-13T01:06:59Z","body":"Thanks, I will definitely try that...\r\nbtw..does the current version support elasticsearch 2.1.x, I found the latest version of ES needs to specify the default index manually, not sure if it's issue\r\n"}}