title: Kafka offset not incremented
body: <p>I am using Kafka with <strong>Spring-boot:</strong> </p>
  
  <p><strong>Kafka Producer class</strong>: </p>
  
  <pre><code>@Service
  public class MyKafkaProducer {
  
      @Autowired
      private KafkaTemplate&lt;String, String&gt; kafkaTemplate;
  
      private static Logger LOGGER = LoggerFactory.getLogger(NotificationDispatcherSender.class);
  
      // Send Message
      public void sendMessage(String topicName, String message) throws Exception {
          LOGGER.debug("========topic Name===== " + topicName + "=========message=======" + message);
          ListenableFuture&lt;SendResult&lt;String, String&gt;&gt; result = kafkaTemplate.send(topicName, message);
          result.addCallback(new ListenableFutureCallback&lt;SendResult&lt;String, String&gt;&gt;() {
              @Override
              public void onSuccess(SendResult&lt;String, String&gt; result) {
                  LOGGER.debug("sent message='{}' with offset={}", message, result.getRecordMetadata().offset());
              }
  
              @Override
              public void onFailure(Throwable ex) {
                  LOGGER.error(Constants.PRODUCER_MESSAGE_EXCEPTION.getValue() + " : " + ex.getMessage());
              }
          });
      }
  }
  </code></pre>
  
  <p><strong>Kafka-configuration:</strong> </p>
  
  <pre><code>spring.kafka.producer.retries=0
  spring.kafka.producer.batch-size=100000
  spring.kafka.producer.request.timeout.ms=30000
  spring.kafka.producer.linger.ms=10
  spring.kafka.producer.acks=0
  spring.kafka.producer.buffer-memory=33554432
  spring.kafka.producer.max.block.ms=5000
  spring.kafka.bootstrap-servers=192.168.1.161:9092,192.168.1.162:9093
  </code></pre>
  
  <p><strong>Problem</strong>: </p>
  
  <p>I have 5 partitions of a topic let's say <code>my-topic</code>. </p>
  
  <p>What happens is, I get success(i.e message is sent to Kafka successfully) logs but offset of none partition of topic <code>my-topic</code> get incremented.</p>
  
  <p>As you can see above, I have added logs <code>onSuccess</code> and <code>onFailure</code>. 
  What I expect is, when Kafka is not able to send a message to Kafka I should get an error, but I do not receive any error message in this case.</p>
  
  <p>The above behavior of Kafka happens at a ratio of <code>100: 5</code> (i.e after every 100 successful message sent to kafka).</p>
  
  <p><strong>Edit1</strong>: Adding Kafka producer logs for a successful case(i.e successfully received message on consumer side)</p>
  
  <pre><code>ProducerConfig - logAll:180] ProducerConfig values: 
      acks = 0
      batch.size = 1000
      block.on.buffer.full = false
      bootstrap.servers = [10.20.1.19:9092, 10.20.1.20:9093, 10.20.1.26:9094]
      buffer.memory = 33554432
      client.id = 
      compression.type = none
      connections.max.idle.ms = 540000
      interceptor.classes = null
      key.serializer = class org.apache.kafka.common.serialization.StringSerializer
      linger.ms = 10
      max.block.ms = 5000
      max.in.flight.requests.per.connection = 5
      max.request.size = 1048576
      metadata.fetch.timeout.ms = 60000
      metadata.max.age.ms = 300000
      metric.reporters = []
      metrics.num.samples = 2
      metrics.sample.window.ms = 30000
      partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
      receive.buffer.bytes = 32768
      reconnect.backoff.ms = 50
      request.timeout.ms = 60000
      retries = 0
      retry.backoff.ms = 100
      sasl.kerberos.kinit.cmd = /usr/bin/kinit
      sasl.kerberos.min.time.before.relogin = 60000
      sasl.kerberos.service.name = null
      sasl.kerberos.ticket.renew.jitter = 0.05
      sasl.kerberos.ticket.renew.window.factor = 0.8
      sasl.mechanism = GSSAPI
      security.protocol = PLAINTEXT
      send.buffer.bytes = 131072
      ssl.cipher.suites = null
      ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
      ssl.endpoint.identification.algorithm = null
      ssl.key.password = null
      ssl.keymanager.algorithm = SunX509
      ssl.keystore.location = null
      ssl.keystore.password = null
      ssl.keystore.type = JKS
      ssl.protocol = TLS
      ssl.provider = null
      ssl.secure.random.implementation = null
      ssl.trustmanager.algorithm = PKIX
      ssl.truststore.location = null
      ssl.truststore.password = null
      ssl.truststore.type = JKS
      timeout.ms = 30000
      value.serializer = class org.apache.kafka.common.serialization.StringSerializer
  
  2017-10-24 14:30:09, [INFO] [karma-unified-notification-manager - ProducerConfig - logAll:180] ProducerConfig values: 
      acks = 0
      batch.size = 1000
      block.on.buffer.full = false
      bootstrap.servers = [10.20.1.19:9092, 10.20.1.20:9093, 10.20.1.26:9094]
      buffer.memory = 33554432
      client.id = producer-1
      compression.type = none
      connections.max.idle.ms = 540000
      interceptor.classes = null
      key.serializer = class org.apache.kafka.common.serialization.StringSerializer
      linger.ms = 10
      max.block.ms = 5000
      max.in.flight.requests.per.connection = 5
      max.request.size = 1048576
      metadata.fetch.timeout.ms = 60000
      metadata.max.age.ms = 300000
      metric.reporters = []
      metrics.num.samples = 2
      metrics.sample.window.ms = 30000
      partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner
      receive.buffer.bytes = 32768
      reconnect.backoff.ms = 50
      request.timeout.ms = 60000
      retries = 0
      retry.backoff.ms = 100
      sasl.kerberos.kinit.cmd = /usr/bin/kinit
      sasl.kerberos.min.time.before.relogin = 60000
      sasl.kerberos.service.name = null
      sasl.kerberos.ticket.renew.jitter = 0.05
      sasl.kerberos.ticket.renew.window.factor = 0.8
      sasl.mechanism = GSSAPI
      security.protocol = PLAINTEXT
      send.buffer.bytes = 131072
      ssl.cipher.suites = null
      ssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]
      ssl.endpoint.identification.algorithm = null
      ssl.key.password = null
      ssl.keymanager.algorithm = SunX509
      ssl.keystore.location = null
      ssl.keystore.password = null
      ssl.keystore.type = JKS
      ssl.protocol = TLS
      ssl.provider = null
      ssl.secure.random.implementation = null
      ssl.trustmanager.algorithm = PKIX
      ssl.truststore.location = null
      ssl.truststore.password = null
      ssl.truststore.type = JKS
      timeout.ms = 30000
      value.serializer = class org.apache.kafka.common.serialization.StringSerializer
  </code></pre>