title: Apache Kafka example error: Failed to send message after 3 tries
body: <p>I am running this kafka producer example mentioned in its site</p>
  
  <p>The code:</p>
  
  <pre><code>public class TestProducer {
  
      public static void main(String[] args) {
          long events = Long.parseLong(args[0]);
          Random rnd = new Random();
          Properties props = new Properties();
          props.put("metadata.broker.list", "host.broker-1:9093, host.broker-2:9093, host.broker-3:9095");
          props.put("serializer.class", "kafka.serializer.StringEncoder");
          props.put("partitioner.class", "test.app.SimplePartitioner");
          props.put("request.required.acks", "1");
          ProducerConfig config = new ProducerConfig(props);
          Producer&lt;String, String&gt; producer = new Producer&lt;String, String&gt;(config);
          for (long nEvents = 0; nEvents &lt; events; nEvents++) { 
                 long runtime = new Date().getTime();  
                 String ip = "192.168.2." + rnd.nextInt(255); 
                 String msg = runtime + ",www.example.com," + ip; 
                 KeyedMessage&lt;String, String&gt; data = new KeyedMessage&lt;String, String&gt;("page_visits", ip, msg);
                 producer.send(data);
          }
          producer.close();
      }
  }
  
  public class SimplePartitioner implements Partitioner{
      public SimplePartitioner (VerifiableProperties props) {
  
      }
  
      public int partition(Object key, int a_numPartitions) {
          int partition = 0;
          String stringKey = (String) key;
          int offset = stringKey.lastIndexOf('.');
          if (offset &gt; 0) {
             partition = Integer.parseInt( stringKey.substring(offset+1)) % a_numPartitions;
          }
         return partition;
    }
  
  }
  </code></pre>
  
  <p>More details:</p>
  
  <p>I am running this application on a host(call is producer) which is remote to host-broker[1-3]</p>
  
  <ul>
  <li><p>I can ping and ssh the broker host from producer host.</p></li>
  <li><p>Provided the advertised.host.name in the server.properties (they are named as server[1-3].properties in the brokers respectively</p></li>
  </ul>
  
  <p>The properties:</p>
  
  <pre><code>broker.id=1
  port=9093
  host.name=host.broker.internal.name
  advertised.host.name=host-broker1
  num.network.threads=3
  num.io.threads=8
  socket.send.buffer.bytes=102400
  socket.receive.buffer.bytes=102400
  socket.request.max.bytes=104857600
  log.dirs=/data/1/kafka-logs-1,/data/2/kafka-logs-2
  num.partitions=1
  num.recovery.threads.per.data.dir=1
  log.retention.hours=168
  log.segment.bytes=1073741824
  log.retention.check.interval.ms=300000
  log.cleaner.enable=false
  zookeeper.connect=zk1:2181,zk2:2181,zk3:2181
  zookeeper.connection.timeout.ms=6000
  </code></pre>
  
  <p>Any idea on how to fix this error?</p>