# Logstash output plugin kafka is not pushing data into kafka

**URL:** <https://discuss.elastic.co/t/logstash-output-plugin-kafka-is-not-pushing-data-into-kafka/104383>\
**Category:** Logstash\
**Created:** [October 18, 2017, 11:44am UTC](https://discuss.elastic.co/t/logstash-output-plugin-kafka-is-not-pushing-data-into-kafka/104383 "2017-10-18T11:44:25Z")\
**Posts on this page:** 5\
**Page:** 1

<div class="post-metadata">

**Author:** ![kabali12345](https://avatars.discourse-cdn.com/v4/letter/k/6a8cbe/32.png) [@kabali12345](https://discuss.elastic.co/u/kabali12345)\
**Post date:** [October 18, 2017, 11:44am UTC](https://discuss.elastic.co/t/logstash-output-plugin-kafka-is-not-pushing-data-into-kafka/104383/1 "2017-10-18T11:44:25Z")

</div>

Hi all  
here i am trying to push Elasticsearch data into kafka

```auto
input { 
  elasticsearch { 
    hosts => "192.168.1.59:9233" 
    index => "titanic1" 
# user => "elastic" 
# password => "changeme" 
# query => '{ "query":{ "match_all": {} }, "sort": ["_doc"] }' 
# query => '{ "query":{"range" : { "@timestamp" : { "gte" : "now-1m/m" } } }, "sort": ["_doc"] }' 
# query => '{ "query":{"range" : { "@timestamp" : { "gte" : "now-1m" } } }, "sort": ["_doc"] }' 
    size => 5000 
    scroll => "5m"
    docinfo => true 
# docinfo_fields => ["_id"]
# docinfo_target => "@metadata" 
} 
 }
#filter {
# mutate {
# add_field => {
# 'doc_id' => "%{[@metadata][_id]}"
# }
# }
#} 
output {
if [Pclass] == "2"
{
kafka {
codec => plain {
format => "%{message}"
 }
# codec => json
bootstrap_servers => ["192.168.1.70:9092"]    
#broker_list => "192.168.1.70:9092"    
topic_id => "sample5"
 }
#exec {
# command => "curl -XDELETE 192.168.1.59:9233/%{[@metadata][_index]}/%{[@metadata][_type]}/%{[@metadata][_id]}?pretty" 
#} 
#stdout {codec => rubydebug } 
#stdout { codec => rubydebug { metadata => true } }
}
} 

```

it is able to contact kafka

```auto
[root@sgplrhel6 ~]# netstat -punta | grep 192.168.1.75
tcp 0 0 ::ffff:192.168.1.70:9092 ::ffff:192.168.1.75:53596 ESTABLISHED 15323/java    

```

and logstash also able to create topic\_name in kafka if not exists  
but logstash failed to push data into topic

here you can see logstash logs

```auto
[2017-10-18T17:13:24,759][DEBUG][logstash.outputs.kafka] KafkaProducer.send() failed: org.apache.kafka.common.errors.TimeoutException: Batch containing 28 record(s) expired due to timeout while requesting metadata from brokers for sample5-0 {:exception=>java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Batch containing 28 record(s) expired due to timeout while requesting metadata from brokers for sample5-0}
[2017-10-18T17:13:24,759][DEBUG][logstash.outputs.kafka] KafkaProducer.send() failed: org.apache.kafka.common.errors.TimeoutException: Batch containing 28 record(s) expired due to timeout while requesting metadata from brokers for sample5-0 {:exception=>java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Batch containing 28 record(s) expired due to timeout while requesting metadata from brokers for sample5-0}
[2017-10-18T17:13:24,760][DEBUG][logstash.outputs.kafka] KafkaProducer.send() failed: org.apache.kafka.common.errors.TimeoutException: Batch containing 28 record(s) expired due to timeout while requesting metadata from brokers for sample5-0 {:exception=>java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Batch containing 28 record(s) expired due to timeout while requesting metadata from brokers for sample5-0}
[2017-10-18T17:13:24,760][DEBUG][logstash.outputs.kafka] KafkaProducer.send() failed: org.apache.kafka.common.errors.TimeoutException: Batch containing 28 record(s) expired due to timeout while requesting metadata from brokers for sample5-0 {:exception=>java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Batch containing 28 record(s) expired due to timeout while requesting metadata from brokers for sample5-0}
[2017-10-18T17:13:24,760][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>9, :failures=>9, :sleep=>0.01}
[2017-10-18T17:13:24,761][DEBUG][logstash.outputs.kafka] KafkaProducer.send() failed: org.apache.kafka.common.errors.TimeoutException: Batch containing 28 record(s) expired due to timeout while requesting metadata from brokers for sample5-0 {:exception=>java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Batch containing 28 record(s) expired due to timeout while requesting metadata from brokers for sample5-0}
[2017-10-18T17:13:24,761][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>13, :failures=>13, :sleep=>0.01}

```

Thank you.

---

<div class="post-metadata">

**Author:** ![dadoonet](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/dadoonet/32/137187_2.png) [@dadoonet](https://discuss.elastic.co/u/dadoonet)\
**Post date:** [October 18, 2017, 11:57am UTC](https://discuss.elastic.co/t/logstash-output-plugin-kafka-is-not-pushing-data-into-kafka/104383/2 "2017-10-18T11:57:47Z")

</div>

I moved your question to #logstash

---

<div class="post-metadata">

**Author:** ![kabali12345](https://avatars.discourse-cdn.com/v4/letter/k/6a8cbe/32.png) [@kabali12345](https://discuss.elastic.co/u/kabali12345)\
**Post date:** [October 18, 2017, 2:27pm UTC](https://discuss.elastic.co/t/logstash-output-plugin-kafka-is-not-pushing-data-into-kafka/104383/3 "2017-10-18T14:27:20Z")

</div>

After updating all logstash plugins  
now i am seeing following logs

```auto
[2017-10-18T19:55:36,794][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>6, :failures=>6, :sleep=>0.01}
[2017-10-18T19:56:06,901][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>1, :failures=>1, :sleep=>0.01}
[2017-10-18T19:56:06,903][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>3, :failures=>3, :sleep=>0.01}
[2017-10-18T19:56:06,905][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>5, :failures=>5, :sleep=>0.01}
[2017-10-18T19:56:06,910][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>6, :failures=>6, :sleep=>0.01}
[2017-10-18T19:56:37,019][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>1, :failures=>1, :sleep=>0.01}
[2017-10-18T19:56:37,029][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>5, :failures=>5, :sleep=>0.01}
[2017-10-18T19:56:37,031][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>3, :failures=>3, :sleep=>0.01}
[2017-10-18T19:56:37,031][INFO][logstash.outputs.kafka] Sending batch to Kafka failed. Will retry after a delay. {:batch_size=>6, :failures=>6, :sleep=>0.01}

```

---

<div class="post-metadata">

**Author:** ![kabali12345](https://avatars.discourse-cdn.com/v4/letter/k/6a8cbe/32.png) [@kabali12345](https://discuss.elastic.co/u/kabali12345)\
**Post date:** [October 21, 2017, 6:42am UTC](https://discuss.elastic.co/t/logstash-output-plugin-kafka-is-not-pushing-data-into-kafka/104383/4 "2017-10-21T06:42:20Z")

</div>

My problem was resolved by adding below line to **server.properties**

```auto
listeners=PLAINTEXT://192.168.1.70:9092

```

[root@sgplrhel6 config]# pwd  
/opt/kafka\_2.10-0.10.0.1/config  
[root@sgplrhel6 config]# ls  
connect-console-sink.properties connect-file-source.properties log4j.properties zookeeper.properties  
connect-console-source.properties connect-log4j.properties producer.properties  
connect-distributed.properties connect-standalone.properties server.properties  
connect-file-sink.properties consumer.properties tools-log4j.properties  
[root@sgplrhel6 config]# vi server.properties  
[root@sgplrhel6 config]#

---

<div class="post-metadata">

**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\
**Post date:** [November 18, 2017, 6:42am UTC](https://discuss.elastic.co/t/logstash-output-plugin-kafka-is-not-pushing-data-into-kafka/104383/5 "2017-11-18T06:42:32Z")

</div>

This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.