Kafkajs: When auto commit false commit offset not working proper way

[jaysequr]

Environment

• Node version : 8.16.2
• Npm version : 6.4.1
• Kafkajs version : 1.11.0

Bug Reproduce Code

• Producer script

const { Kafka } = require('kafkajs');
const kafka = new Kafka({
    clientId: 'my-app',
    brokers: ['aws.confluent.cloud:9092'],
    ssl: true,
    sasl: {
      mechanism: 'plain', // scram-sha-256 or scram-sha-512
      username: 'confluent_username',
      password: 'confluent_password'
    },
});
const topic_name = 'demo-topic';
const producer = kafka.producer();
const initialize_producer_connection = async ()=>{

    await producer.connect();

    await producer.send({
        topic: topic_name,
        messages: [
            { key: 'key1', value: 'hello world 1' },
            { key: 'key2', value: 'hello world 2' },
            { key: 'key3', value: 'hello world 3' },
            { key: 'key4', value: 'hello world 4' },
            { key: 'key5', value: 'hello world 5' },
            { key: 'key6', value: 'hello world 6' },
        ],
    });

};
initialize_producer_connection();

• Consumer script

const { Kafka } = require('kafkajs');
const kafka = new Kafka({
    clientId: 'my-app',
    brokers: ['aws.confluent.cloud:9092'],
    ssl: true,
    sasl: {
      mechanism: 'plain', // scram-sha-256 or scram-sha-512
      username: 'confluent_username',
      password: 'confluent_password'
    },
});
const topic_name = 'demo-topic';
const group_id = 'TESTING_GROUP'; 
const consumer = kafka.consumer({
    groupId: group_id,
});
const initialize_consumer_connection = async ()=>{

    await admin.connect();

    await consumer.connect();

    await consumer.subscribe({ 
        topic: topic_name, 
        fromBeginning: true 
    });

    await consumer.run({
        autoCommit:false,
        eachMessage: async ({ topic, partition, message }) => {

            const prefix = `${topic}[${partition} | ${message.offset}] / ${message.timestamp}`
            console.log(`- ${prefix} ${message.key}#${message.value}`);

            const commit_message = {
                topic,
                partition,
                offset: message.offset
            };

            consumer.commitOffsets([commit_message]);

        },
    });

};
initialize_consumer_connection();

Issue Description

• When starting consumer script old message received which has already consume that message in past.
• I try to manual commit to messaging so it auto-commit false in consumer-run.
• Can I achieve manual commit of this way?

[Vuksan]

I think that you need to call consumer.commitOffsets function after consumer.run, as stated Here.

If you want to manually commit, with auto-commit turned off, and you want to do it inside the consumer.run function, you need to use a combination of eachBatch and commitOffsetsIfNecessary.

Disclamer: I am not involved with kafkajs in any way, so I might be wrong.
I am just trying to help before someone from the team finds time. 🙂

[andycmaj]

yeah the documentation is a bit hazy here IMO:

Committing offsets does not change what message we'll consume next once we've started consuming, but instead is only used to determine from which place to start. To immediately change from what offset you're consuming messages, you'll want to seek, instead.

that COULD imply that this is the expected behavior.

another thing that's confusing is the algo for their suggested implementation of tracking your own offsets:

The usual usage pattern for offsets stored outside of Kafka is as follows:

Run the consumer with autoCommit disabled.
Store a message's offset + 1 in the store together with the results of processing. 1 is added to prevent that same message from being consumed again.
Use the externally stored offset on restart to seek the consumer to it.

note that they suggest adding 1 to prevent the same message from being consumed again... so maybe they expect you to do this yourself?

unclear.

anyway i was experiencing the same issue, and i did something like this:

1. connect consumer
2. get latest offset from Admin client
3. run consumer.
4. for each message received, check if the offset is <= results from 2. if so, skip this message.

To be clear, the commit offsets ARE updated correctly, it's just that the consumer starts consuming AT the latest offset, NOT latest offset + 1.

this seems counter-intuitive, but kinda documented... weirdly?

[andycmaj]

@tulios are there any other possible for dupes or causes for this off-by-one issue? or are we just making invalid assumptions. if the latter, i think we have a documentation bug here at the least :)

[rhyek]

For what it's worth, I tested setting both offset and offset + 1 with process restarts and this is the correct way to do this:

await consumer.run({
  autoCommit: false,
  eachMessage: async ({ topic, partition, message }) => {
    ...
    await consumer.commitOffsets([{ topic, partition, offset: (Number(message.offset) + 1).toString() }]);
  },
});

with just await consumer.commitOffsets([{ topic, partition, offset: message.offset }]); it will re-process the last message after process restart.

If there are any drawbacks to this it would be great to hear from the devs.

[andycmaj]

🤔 my first thought was that it was a bit counter-intuitive... if offset means, "the last message successfully processed".

if offset means, "the message at which consumption will start when a new consumer starts listening", then it makes more sense, but i don't think that's Kafka's intended semantic.

• https://www.confluent.io/blog/apache-kafka-data-access-semantics-consumers-and-membership/
  > The OffsetCommitRequest consists of a map that denotes the latest processed offset for any given partition (TopicPartition->OffsetAndMetadata).

I believe this is a common confusion in kafka client libraries. see this python issue and this one.

seems like a mismatch between commit and consume semantics.

commit clearly (from confluent/kafka docs) should be dealing with the last processed message.

i believe CONSUMERS should start processing from the last committed message + 1 when they start up.

however there are other libraries that seem to confuse the issue:

• https://github.com/edenhill/librdkafka/wiki/Consumer-offset-management
  > Commit - Offset committed to permanent storage (broker, file). When consumer restarts this is where it will start consuming from. The committed offset should be last_message_offset+1.

it's unclear to me whether they mean that the consumer will handle this or that the user should manually commit last_message_offset + 1.

again. personally, based on kafka docs, "The OffsetCommitRequest consists of a map that denotes the latest processed offset", i believe that the "fetch next offset after last committed one" should be the responsibility of consumer join internals.

[rhyek]

Yes, I agree with you. The way I had understood it was the developer should commit the offset already processed, not +1. Nevertheless, I assume this is the way KafkaJS will continue to handle it to avoid breaking current usage. The documentation should make this detail clear.

[edmondo1984]

@rhyek are you saying that we should commit offset + 1?