kafka-node: Wrong offset is being committed, messages are missing

@hyperlink thanks again for your support and attempts to ensure the trust in this lib, we were able to identify the reason for the problems. And I am happy to sum them up, for anyone else who might be stuck in a similar situation in the future.

1. using 1 or 30 partitions does not matter, having a single instance or a cluster does not matter either, we ran the large-message-test against all kinds of setups.

2. combining .setOffset with manual consumer.commits() is a bad idea

3. expecting .setOffset(topic, partition, 0) to work on a running consumer, even paused, can be problematic and is probably a bad behavior for any kind of kafka consumer

4. when “experiencing missing messages” and “committed high offsets” make sure to understand the topic you are consuming - we have a large amount of topics here, with different owners. Keyed partitions might be very well subject to log compaction in large fashion depending on the data and frequency of publishes the topic goes through.

@krystianity here’s the updated test I wrote earlier to consume from 30 partitions w/ autoCommit off. I ran it several times without issues.

I ran the test several times in a row (test need to be adjusted) without resetting the kafka container was able to consume up to 6.3 million messages in 527 seconds.

Setup topic

Kafka bin tools

./kafka-topics.sh --zookeeper localhost:2181 --create --replication-factor 1 --partitions 30 --topic ThirtyPartitions

Test code

./test/large-cs-test.js

'use strict';

const host = process.env['KAFKA_TEST_HOST'] || '';
const ConsumerGroup = require('../lib/consumerGroup');
const _ = require('lodash');
const uuid = require('uuid');
const async = require('async');
const Client = require('../lib/client');
const Producer = require('../lib/highLevelProducer');
const assert = require('assert');

let topic, client, consumerGroup, consumed, interval;
const numberOfMessages = 2100000;
const chunkSize = 30;

function sendUUIDMessages (times, topic, done) {
  console.log(`Trying to send ${times} messages`);
  const producer = new Producer(client, { requireAcks: 1 });
  assert(producer.ready, 'Producer is not ready');
  console.log('producer ready');
  const allMessages = _.times(times, uuid.v4);
  const chunked = _.chunk(allMessages, times / chunkSize);
  console.log(`${allMessages.length} messages generated. Sending in ${chunked.length} chunks of ${times / chunkSize}`);
  let count = 0;
  async.eachSeries(chunked, function (messages, callback) {
    console.log(`sending ${++count}`);
    producer.send([{topic: topic, messages: messages}], callback);
  }, done);
  return producer;
}

function init (done) {
  topic = 'ThirtyPartitions'; // uuid.v4();
  async.series([
    function (callback) {
      client = new Client(host, uuid.v4());
      client.once('connect', callback);
    },
    function (callback) {
      client.refreshMetadata([topic], callback);
    },
    function (callback) {
      sendUUIDMessages(numberOfMessages, topic, function (error) {
        if (error) {
          return callback(error);
        }
        assert.equal(client.topicPartitions[topic].length, chunkSize);
        callback();
      });
    },
    function (callback) {
      client.close(callback);
    }
  ], function (error) {
    if (error) {
      return done(error);
    }
    console.log('All messages sent');
    done();
  });
}

function bench (done) {
  const groupId = uuid.v4();
  console.log(`starting to consume using groupId: ${groupId}`);
  const time = process.hrtime();
  consumerGroup = new ConsumerGroup({
    fetchMaxBytes: 1024 * 100,
    groupId: groupId,
    host: host,
    sessionTimeout: 8000,
    heartbeatInterval: 7000,
    retryMinTimeout: 250,
    autoCommit: false,
    fromOffset: 'earliest'
  }, topic);

  const stats = {}

  consumed = 0;
  consumerGroup.once('error', done);
  consumerGroup.on('message', function (message) {
    stats[message.partition] = message.offset;

    if (++consumed === numberOfMessages) {
      console.log(`consumed ${consumed}`);
      const [seconds, nanoseconds] = process.hrtime(time);
      console.log(`took ${seconds}s ${nanoseconds / 1e6}ms`);
      console.log(stats);
      clearInterval(interval);
      done();
    }
  });

  consumerGroup.once('connect', function () {
    interval = setInterval(function () {
      console.log(`consumed ${consumed}`);
    }, 1000);
  });
}

async.series([init, bench], function (error) {
  if (error) {
    console.error('failed with', error);
  }
  client.close();
  consumerGroup.close(function() {})
});

I wrote a simple test single consumer against a single broker, and partition. I published 2.1 million messages and was able to consume them without an issue. Ran it several times. https://github.com/SOHU-Co/kafka-node/commit/283e6a88492168ed86993b683e5fc0868a469e26

If you’re interested in running it checkout the large-message-test branch.

You will need to run the test w/o timeout since it could take about 14 minutes to run.

./node_modules/.bin/mocha --no-timeouts test/test.consumerGroup.js --grep "large"

How many consumers are your group?