title: Not able to send json tweets events to Kafka topic/producer using kafka command line
body: <p>I have created a python script <code>raw_tweets_stream.py</code> to stream twitter data using twitter api. The json data from twitter is pipped to kafka producer using the script below.</p>
  
  <hr>
  
  <pre><code>`python raw_tweets_stream.py | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list localhost:2181 --topic raw_json_tweets`
  </code></pre>
  
  <hr>
  
  <p><code>raw_json_tweets</code> is the kafka topic created for these tweets. The python script <code>raw_tweets_stream.py</code> runs just fine but it throws error while sending it to the kafka producer. I am using Hortonworks HDP 2.3.1 sandbox and I have made sure that the zookeeper and kafka are started. </p>
  
  <hr>
  
  <p><code>/usr/hdp/current/kafka-broker/bin/kafka-topics.sh --zookeeper localhost:2181 --describe --topic raw_json_tweets</code></p>
  
  <pre><code>Topic:raw_json_tweets      PartitionCount:1        ReplicationFactor:1     Configs:
              Topic: raw_json_tweets     Partition: 0    Leader: 0       Replicas: 0     Isr: 0
  </code></pre>
  
  <p>Error:</p>
  
  <pre><code>[2016-08-25 22:36:26,212] ERROR Failed to send requests for topics raw_json_tweets with correlation ids in [57,64] (kafka.producer.async.DefaultEventHandler)
  [2016-08-25 22:36:26,213] ERROR Error in handling batch of 131 events (kafka.producer.async.ProducerSendThread)
  kafka.common.FailedToSendMessageException: Failed to send messages after 3 tries.
          at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:91)
          at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105)
          at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:88)
          at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:68)
          at scala.collection.immutable.Stream.foreach(Stream.scala:547)
          at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
          at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
  [2016-08-25 22:36:27,217] WARN Fetching topic metadata with correlation id 65 for topics [Set(json_tweets1)] from broker [BrokerEndPoint(0,localhost,2181)] failed (kafka.client.ClientUtils$)
  java.io.EOFException: Received -1 when reading from channel, socket has likely been closed.
          at kafka.utils.CoreUtils$.read(CoreUtils.scala:193)
          at kafka.network.BoundedByteBufferReceive.readFrom(BoundedByteBufferReceive.scala:54)
          at kafka.network.Receive$class.readCompletely(Transmission.scala:56)
          at kafka.network.BoundedByteBufferReceive.readCompletely(BoundedByteBufferReceive.scala:29)
          at kafka.network.BlockingChannel.receive(BlockingChannel.scala:131)
          at kafka.producer.SyncProducer.liftedTree1$1(SyncProducer.scala:77)
          at kafka.producer.SyncProducer.kafka$producer$SyncProducer$$doSend(SyncProducer.scala:74)
          at kafka.producer.SyncProducer.send(SyncProducer.scala:115)
          at kafka.client.ClientUtils$.fetchTopicMetadata(ClientUtils.scala:59)
          at kafka.producer.BrokerPartitionInfo.updateInfo(BrokerPartitionInfo.scala:82)
          at kafka.producer.BrokerPartitionInfo.getBrokerPartitionInfo(BrokerPartitionInfo.scala:49)
          at kafka.producer.async.DefaultEventHandler.kafka$producer$async$DefaultEventHandler$$getPartitionListForTopic(DefaultEventHandler.scala:188)
          at kafka.producer.async.DefaultEventHandler$$anonfun$partitionAndCollate$1.apply(DefaultEventHandler.scala:152)
          at kafka.producer.async.DefaultEventHandler$$anonfun$partitionAndCollate$1.apply(DefaultEventHandler.scala:151)
          at scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59)
          at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47)
          at kafka.producer.async.DefaultEventHandler.partitionAndCollate(DefaultEventHandler.scala:151)
          at kafka.producer.async.DefaultEventHandler.dispatchSerializedData(DefaultEventHandler.scala:96)
          at kafka.producer.async.DefaultEventHandler.handle(DefaultEventHandler.scala:73)
          at kafka.producer.async.ProducerSendThread.tryToHandle(ProducerSendThread.scala:105)
          at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:88)
          at kafka.producer.async.ProducerSendThread$$anonfun$processEvents$3.apply(ProducerSendThread.scala:68)
          at scala.collection.immutable.Stream.foreach(Stream.scala:547)
          at kafka.producer.async.ProducerSendThread.processEvents(ProducerSendThread.scala:67)
          at kafka.producer.async.ProducerSendThread.run(ProducerSendThread.scala:45)
  </code></pre>
  
  <hr>
  
  <p><strong>Update: Solution</strong></p>
  
  <hr>
  
  <ol>
  <li>Went to Ambari Services and changed the Kafka logs directory to <code>/tmp/kafka-logs</code>.</li>
  <li><p>Modified the original script to include the correct port and hostname. </p>
  
  <p><code>python raw_tweets_stream.py | /usr/hdp/current/kafka-broker/bin/kafka-console-producer.sh --broker-list sandbox.hortonworks.com:6667 --topic raw_json_tweets</code></p></li>
  <li><p>Verified that the events are sent to the kafka topic using console consumer.</p>
  
  <p><code>/usr/hdp/2.3.0.0-2557/kafka/bin/kafka-console-consumer.sh -zookeeper sandbox.hortonworks.com:2181 -topic raw_json_tweets -from-beginning</code></p></li>
  </ol>