title: Kafka Consumer API jumping offsets
body: <p>I am using Kafka Version 2.0 and java consumer API to consume messages from a topic. We are using a single node Kafka server with one consumer per partition. I have observed that the consumer is loosing some of the messages.
  The scenario is:
  Consumer polls the topic.
  I have created One Consumer Per Thread.
  Fetches the messages and gives it to a handler to handle the message.
  Then it commits the offsets using "At-least-once" Kafka Consumer semantics to commit Kafka offset.
  In parallel, I have another consumer running with a different group-id. In this consumer, I'm simply increasing the message counter and committing the offset. There's no message loss in this consumer.   </p>
  
  <pre><code>try {
      //kafkaConsumer.registerTopic();
  
      consumerThread = new Thread(() -&gt; {
          final String topicName1 = "topic-0";
          final String topicName2 = "topic-1";
          final String topicName3 = "topic-2";
          final String topicName4 = "topic-3";
  
          String groupId = "group-0";
          final Properties consumerProperties = new Properties();
          consumerProperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.13.49:9092");
          consumerProperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
          consumerProperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.ByteArrayDeserializer");
          consumerProperties.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
          consumerProperties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "100");
          consumerProperties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
          consumerProperties.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 1000);
  
          try {
              consumer = new KafkaConsumer&lt;&gt;(consumerProperties);
              consumer.subscribe(Arrays.asList(topicName1, topicName2, topicName3, topicName4));
          } catch (KafkaException ke) {
              logTrace(MODULE, ke);
          }
          while (service.isServiceStateRunning()) {
              ConsumerRecords&lt;String, byte[]&gt; records = consumer.poll(Duration.ofMillis(100));
              for (TopicPartition partition : records.partitions()) {
                  List&lt;ConsumerRecord&lt;String, byte[]&gt;&gt; partitionRecords = records.records(partition);
                  for (ConsumerRecord&lt;String, byte[]&gt; record : partitionRecords) {
                      processMessage(simpleMessage);
  
                  }
              }
              consumer.commitSync();
          }
          kafkaConsumer.closeResource();
      }, "KAKFA_CONSUMER");
  
  } catch (Exception e) {
  }
  </code></pre>