title: Delay in Consumer consuming messages in Apache Kafka
body: <p>I am using Kafka 0.8.0 and trying to achieve the below mentioned scenario.</p>
  
  <p><strong>JCA API (Acts as a producer and sends data to)-----> Consumer------> HBase</strong></p>
  
  <p>I am sending each message to consumer as soon as I fetch the data using JCA Client. For instance, as soon as producer sends message no.1 , I want to fetch the same from consumer and 'put' in HBase. But my consumer starts fetching the messages after some random n messages . I want to put the producer and consumer in sync so that both of them start working together.</p>
  
  <p>I have used:</p>
  
  <p><strong>1 broker</strong></p>
  
  <p><strong>1 single topic</strong></p>
  
  <p><strong>1 single producer and high level Consumer</strong></p>
  
  <p>Can anyone suggest what do i need to do to achieve the same?</p>
  
  <p>EDITED:</p>
  
  <p>Adding some relevant code snippet.</p>
  
  <p><strong>Consumer.java</strong></p>
  
  <pre><code>public class Consumer extends Thread {
      private final ConsumerConnector consumer;
      private final String topic;
      PrintWriter pw = null;
      int t = 0;
      StringDecoder kd = new StringDecoder(null);
      Map&lt;String, Integer&gt; topicCountMap = new HashMap&lt;String, Integer&gt;();
      Map&lt;String, List&lt;KafkaStream&lt;String, Signal&gt;&gt;&gt; consumerMap;
      KafkaStream&lt;String, Signal&gt; stream;
      ConsumerIterator&lt;String, Signal&gt; it;
  
      public Consumer(String topic) {
          consumer = kafka.consumer.Consumer.createJavaConsumerConnector(createConsumerConfig());
  
          this.topic = topic;
          topicCountMap.put(topic, new Integer(1));
          consumerMap = consumer.createMessageStreams(topicCountMap, kd, new Serializer(
                  new VerifiableProperties()));
          stream = consumerMap.get(topic).get(0);
          it = stream.iterator();
  
      }
  
      private static ConsumerConfig createConsumerConfig() {
          Properties props = new Properties();
          props.put("zookeeper.connect", KafkaProperties.zkConnect);
          props.put("group.id", KafkaProperties.groupId);
          props.put("zookeeper.session.timeout.ms", "400");
          props.put("zookeeper.sync.time.ms", "200");
          props.put("auto.commit.interval.ms", "1000");
          props.put("fetch.size", "1024");
  
          return new ConsumerConfig(props);
  
      }
  
      synchronized public void run() {
  
          while (it.hasNext()) {
              t = (it.next().message()).getChannelid();
              System.out.println("In Consumer received msg" + t);
          }
      }
  }
  </code></pre>
  
  <p><strong>producer.java</strong></p>
  
  <pre><code>public class Producer {
      public final kafka.javaapi.producer.Producer&lt;String, Signal&gt; producer;
      private final String topic;
      private final Properties props = new Properties();
  
      public Producer(String topic)
      {
          props.put("serializer.class", "org.bigdata.kafka.Serializer");
          props.put("key.serializer.class", "kafka.serializer.StringEncoder");
          props.put("metadata.broker.list", "localhost:9092");
          // Use random partitioner. Don't need the key type. Just set it to Integer.
          // The message is of type userdefined Object .
          producer = new kafka.javaapi.producer.Producer&lt;String,Signal(newProducerConfig(props));
          this.topic = topic;
      }
  }
  </code></pre>
  
  <p><strong>KafkaProperties.java</strong></p>
  
  <pre><code>public interface KafkaProperties {
      final static String zkConnect = "127.0.0.1:2181";
      final static String groupId = "group1";
      final static String topic = "test00";
      final static String kafkaServerURL = "localhost";
      final static int kafkaServerPort = 9092;
      final static int kafkaProducerBufferSize = 64 * 1024;
      final static int connectionTimeOut = 100000;
      final static int reconnectInterval = 10000;
      final static String clientId = "SimpleConsumerDemoClient";
  }
  </code></pre>
  
  <p><strong>This is how the consumer is behaving for the first 10 messages it does not sysout that message received by consumer but from the 11th message onwards it starts functioning correctly.</strong></p>
  
  <pre><code>     producer sending msg1
  
       producer sending msg2
  
       producer sending msg3
  
       producer sending msg4
  
       producer sending msg5
  
       producer sending msg6
  
       producer sending msg7
  
       producer sending msg8
  
       producer sending msg9
  
       producer sending msg10
  
       producer sending msg11
  
       producer sending msg12
       In Consumer received msg12
  
       producer sending msg13
       In Consumer received msg13
  
       producer sending msg14
       In Consumer received msg14
  
       producer sending msg15
       In Consumer received msg15
  
       producer sending msg16
       In Consumer received msg16
  
       producer sending msg17
       In Consumer received msg17
  
       producer sending msg18
       In Consumer received msg18
  
       producer sending msg19
       In Consumer received msg19
  
       producer sending msg20
       In Consumer received msg20
  
       producer sending msg21
       In Consumer received msg21
  </code></pre>
  
  <p><strong>EDITED: adding the listener function where producer is sending messages to consumer. And I am using the default producer config did not overwrite it</strong></p>
  
  <pre><code>public synchronized void onValueChanged(final MonitorEvent event_) {
  
  
      // Get the value from the DBR
      try {
          final DBR dbr = event_.getDBR();
  
          final String[] val = (String[]) dbr.getValue();
  
          producer1.producer.send(new KeyedMessage&lt;String, Signal&gt;         
                      (KafkaProperties.topic,new Signal(messageNo)));
          System.out.println("producer sending msg"+messageNo);
  
          messageNo++;
  
  
      } catch (Exception ex) {
          ex.printStackTrace();
      }
  }
  </code></pre>