Kafkajs: fetchOffsets after resetOffsets returns -1 for all partitions

Created on 3 Nov 2018  路  9Comments  路  Source: tulios/kafkajs

It seems fetchOffsets after resetOffsets keeps returning -1 to all partitions, even though there are previously stored offsets.

In principle, what I would like to do, is to output the current offsets to logs when a new consumer is added. In our use case, we should always start from the latest offset.

Here's the code I used for testing (kafkajs version 1.4.4, kafka version 1.0.1):

const client = new kafka.Kafka({
    brokers: ['127.0.0.1:9092'],
    clientId: 'test-clientId'
});

const admin = client.admin();
await admin.connect();
await admin.resetOffsets({ groupId: 'test-groupId', topic: 'test-topic' });
// ALL PARTITION OFFSETS EQUAL -1
console.log(
    'FETCH OFFSETS',
    await admin.fetchOffsets({
        groupId: 'test-groupId',
        topic: 'test-topic'
    })
);

const consumer = client.consumer({
    groupId: 'test-groupId'
});
await consumer.subscribe({ topic: 'test-topic' });
await consumer.connect();
await consumer.run({
    eachMessage: async ({ message, partition }) => {
        console.log('RECEIVED MESSAGE', partition, message.offset);
    }
});

const producer = client.producer();
await producer.connect();
await producer.send({
    messages: [{ key: '', value: 'Hello world' }],
    topic: 'test-topic'
});
await producer.disconnect();
await consumer.disconnect();

await admin.resetOffsets({ groupId: 'test-groupId', topic: 'test-topic' });
// ALL PARTITION OFFSETS STILL EQUAL -1
console.log(
    'FETCH OFFSETS',
    await admin.fetchOffsets({
        groupId: 'test-groupId',
        topic: 'test-topic'
    })
);

await admin.disconnect();

If it helps, here's the stdout after running the previous code block:

{"level":"INFO","timestamp":"2018-11-03T09:37:01.697Z","logger":"kafkajs","message":"[Consumer] Starting","groupId":"test-groupId"}
{"level":"INFO","timestamp":"2018-11-03T09:37:01.698Z","logger":"kafkajs","message":"[ConsumerGroup] Pausing fetching from 1 topics","topics":["test-topic"]}
{"level":"INFO","timestamp":"2018-11-03T09:37:01.712Z","logger":"kafkajs","message":"[Runner] Consumer has joined the group","groupId":"test-groupId","memberId":"test-clientId-70e47b46-abb4-4ce9-bc69-7b60c289d4ee","leaderId":"test-clientId-70e47b46-abb4-4ce9-bc69-7b60c289d4ee","isLeader":true,"memberAssignment":{"test-topic":[2,1,0]},"duration":14}
{"level":"INFO","timestamp":"2018-11-03T09:37:07.748Z","logger":"kafkajs","message":"[Consumer] Stopped","groupId":"test-groupId"}
FETCH OFFSETS [ { partition: 0, offset: '-1' },
  { partition: 1, offset: '-1' },
  { partition: 2, offset: '-1' } ]
{"level":"INFO","timestamp":"2018-11-03T09:37:07.759Z","logger":"kafkajs","message":"[Consumer] Starting","groupId":"test-groupId"}
{"level":"INFO","timestamp":"2018-11-03T09:37:07.767Z","logger":"kafkajs","message":"[Runner] Consumer has joined the group","groupId":"test-groupId","memberId":"test-clientId-91a4b8ee-da97-49c8-afae-c0db0c71f70e","leaderId":"test-clientId-91a4b8ee-da97-49c8-afae-c0db0c71f70e","isLeader":true,"memberAssignment":{"test-topic":[2,1,0]},"duration":8}
RECEIVED MESSAGE 0 15
{"level":"INFO","timestamp":"2018-11-03T09:37:08.821Z","logger":"kafkajs","message":"[Consumer] Stopped","groupId":"test-groupId"}
{"level":"INFO","timestamp":"2018-11-03T09:37:08.824Z","logger":"kafkajs","message":"[Consumer] Starting","groupId":"test-groupId"}
{"level":"INFO","timestamp":"2018-11-03T09:37:08.824Z","logger":"kafkajs","message":"[ConsumerGroup] Pausing fetching from 1 topics","topics":["test-topic"]}
{"level":"INFO","timestamp":"2018-11-03T09:37:08.830Z","logger":"kafkajs","message":"[Runner] Consumer has joined the group","groupId":"test-groupId","memberId":"test-clientId-d2bd49d8-31f4-4561-af03-41adabf9be62","leaderId":"test-clientId-d2bd49d8-31f4-4561-af03-41adabf9be62","isLeader":true,"memberAssignment":{"test-topic":[2,1,0]},"duration":6}
{"level":"INFO","timestamp":"2018-11-03T09:37:14.846Z","logger":"kafkajs","message":"[Consumer] Stopped","groupId":"test-groupId"}
FETCH OFFSETS [ { partition: 0, offset: '-1' },
  { partition: 1, offset: '-1' },
  { partition: 2, offset: '-1' } ]
question
Source
picture kpala

All 9 comments

@kpala I think you are not giving enough time for your consumers to process the message, you are sending the message and disconnecting right after it. Another thing, the only way you have to know when the consumer has joined the group is listening to the instrumentation event GROUP_JOIN, something like:

const { GROUP_JOIN } = consumer.events
consumer.run({ ... }
// ...
consumer.on(GROUP_JOIN, async event => {
  console.log(e.payload. memberAssignment)
})
tulios picture tulios on 5 Nov 2018
👍1

Thanks @tulios for your ideas! I now added a timeout before disconnecting and tried fetching offsets at three different times:

  1. After GROUP_JOIN event -> Offsets for all partitions equal -1
  2. After message was received (and after waiting for 10sec) -> The partition where the message exists, has now offset greater than -1
  3. After resetOffsets -> Offsets for all partitions equal -1, again

It seems as if fetchOffsets is not properly refreshing metadata from the Kafka brokers. (Or is it just intentional, that "latest" offsets equal -1?)

Furthermore, on GROUP_JOIN event, event.payload.memberAssignment seems to contain only partitions for which the consumer group was assigned to, but not the corresponding offsets. Not sure if the offsets should be there at all though?

Here's the modified code used for testing:

const client = new kafka.Kafka({
    brokers: ['127.0.0.1:9092'],
    clientId: 'test-clientId'
});

const admin = client.admin();
await admin.connect();

const consumer = client.consumer({
    groupId: 'test-groupId'
});
await consumer.subscribe({ topic: 'test-topic' });
await consumer.connect();
consumer.on(consumer.events.GROUP_JOIN, async event => {
    console.log('GROUP JOIN', event.payload.memberAssignment);
    // ALL PARTITION OFFSETS EQUAL -1
    console.log(
        'FETCH OFFSETS AFTER GROUP JOIN',
        await admin.fetchOffsets({
            groupId: 'test-groupId',
            topic: 'test-topic'
        })
    );
});
await consumer.run({
    eachMessage: async ({ message, partition }) => {
        console.log('RECEIVED MESSAGE', partition, message.offset);
    }
});

const producer = client.producer();
await producer.connect();
await producer.send({
    messages: [{ key: null, value: 'Hello world' }],
    topic: 'test-topic'
});

await new Promise(resolve => {
    setTimeout(resolve, 10000);
});
// ONE PARTITION HAS NOW OFFSET, OTHERS ARE STILL -1
console.log(
    'FETCH OFFSETS AFTER MESSAGE RECEIVED + TIMEOUT',
    await admin.fetchOffsets({
        groupId: 'test-groupId',
        topic: 'test-topic'
    })
);

await consumer.disconnect();
await producer.disconnect();

await admin.resetOffsets({ groupId: 'test-groupId', topic: 'test-topic' });

await new Promise(resolve => {
    setTimeout(resolve, 10000);
});
// ALL PARTITION OFFSETS EQUAL -1 AGAIN
console.log(
    'FETCH OFFSETS AFTER RESET OFFSETS',
    await admin.fetchOffsets({
        groupId: 'test-groupId',
        topic: 'test-topic'
    })
);

await admin.disconnect();

And here's the stdout:

{"level":"INFO","timestamp":"2018-11-05T14:06:16.223Z","logger":"kafkajs","message":"[Consumer] Starting","groupId":"test-groupId"}
GROUP JOIN { 'test-topic': [ 2, 1, 0 ] }
{"level":"INFO","timestamp":"2018-11-05T14:06:16.245Z","logger":"kafkajs","message":"[Runner] Consumer has joined the group","groupId":"test-groupId","memberId":"test-clientId-abc8ccec-9888-4184-8773-de2827ecc708","leaderId":"test-clientId-abc8ccec-9888-4184-8773-de2827ecc708","isLeader":true,"memberAssignment":{"test-topic":[2,1,0]},"duration":20}
FETCH OFFSETS AFTER GROUP JOIN [ { partition: 0, offset: '-1' },
  { partition: 1, offset: '-1' },
  { partition: 2, offset: '-1' } ]
RECEIVED MESSAGE 2 0
FETCH OFFSETS AFTER MESSAGE RECEIVED + TIMEOUT [ { partition: 0, offset: '-1' },
  { partition: 1, offset: '-1' },
  { partition: 2, offset: '1' } ]
{"level":"INFO","timestamp":"2018-11-05T14:06:27.324Z","logger":"kafkajs","message":"[Consumer] Stopped","groupId":"test-groupId"}
FETCH OFFSETS AFTER CONSUMER DISCONNECT [ { partition: 0, offset: '-1' },
  { partition: 1, offset: '-1' },
  { partition: 2, offset: '1' } ]
{"level":"INFO","timestamp":"2018-11-05T14:06:27.337Z","logger":"kafkajs","message":"[Consumer] Starting","groupId":"test-groupId"}
{"level":"INFO","timestamp":"2018-11-05T14:06:27.337Z","logger":"kafkajs","message":"[ConsumerGroup] Pausing fetching from 1 topics","topics":["test-topic"]}
{"level":"INFO","timestamp":"2018-11-05T14:06:27.344Z","logger":"kafkajs","message":"[Runner] Consumer has joined the group","groupId":"test-groupId","memberId":"test-clientId-eeda10be-0e38-46e3-b5b7-c84386beedfc","leaderId":"test-clientId-eeda10be-0e38-46e3-b5b7-c84386beedfc","isLeader":true,"memberAssignment":{"test-topic":[2,1,0]},"duration":7}
{"level":"INFO","timestamp":"2018-11-05T14:06:33.374Z","logger":"kafkajs","message":"[Consumer] Stopped","groupId":"test-groupId"}
FETCH OFFSETS AFTER RESET OFFSETS [ { partition: 0, offset: '-1' },
  { partition: 1, offset: '-1' },
  { partition: 2, offset: '-1' } ]
kpala picture kpala on 5 Nov 2018

You are right, resetOffsets basically set the offsets to the special values for earliest or latest (-1 or -2), then as the consumers start they eventually resolve the offsets into real values.

About the instrumentation events, take a look here:
https://github.com/tulios/kafkajs#instrumentation

but basically, you can get the information you want from start or end batch process

consumer.events.START_BATCH_PROCESS
payload: {topic, partition, highWatermark, offsetLag, batchSize, firstOffset, lastOffset}

consumer.events.END_BATCH_PROCESS
payload: {topic, partition, highWatermark, offsetLag, batchSize, firstOffset, lastOffset, duration}
tulios picture tulios on 5 Nov 2018

Ahh that explains it then!

I'm considering a switch from kafka-node to your library (due to numerous reasons) and I of course assumed fetching would work in a similar manner as https://github.com/SOHU-Co/kafka-node#fetchpayloads-cb or https://github.com/SOHU-Co/kafka-node#fetchcommitsv1groupid-payloads-cb

The suggestion for using START/END_BATCH_PROCESS events would work to some extent but they require that the consumer receives a batch of messages first whereas I'd like to know the committed offsets before any messages are received. Furthermore, these events provide information for a single partition only.

In any case, I appreciate your help so far. I can live without the feature I'm after for now. If such feature fits to your roadmap, would be great though!

kpala picture kpala on 5 Nov 2018

Nice
One question, why do you need to reset the offsets before issuing fetchOffsets?
fetchOffsets should return what you need but maybe I'm missing something.

tulios picture tulios on 5 Nov 2018

Sure, maybe I should indeed elaborate a little bit.

We have a use case where the consumer needs to always start from the latest offset. In principle, we have two devices that communicate with each other, but the communication is initiated by the end user in real time. The end user may restart any of the devices at any time (e.g. due to problems currently resolvable by rebooting). When the device comes back online, we don't want to process any messages that may have been sent to Kafka in the meanwhile. This is because a message may trigger a lengthy run on the receiving device and cancelling the run is tedious and requires the end user to be present.

We've also seen cases in production, where some messages have been received and processed but committing the offsets back has failed. Then, when the consumer restarts (automatically or triggered by the end user), the same messages are processed again against our will.

This is why we need to reset the offsets before starting to consume any messages. (Alternatively, we could use a different consumer groupId every time the consumer is started but it's not that elegant solution.) Finally, why we are interested in outputting the latest committed offsets, is to make sure that the consumer is really started from the latest position.

As a reference, kafka-node also has a convenience method "fetchLatestOffsets" https://github.com/SOHU-Co/kafka-node#fetchlatestoffsetstopics-cb

Maybe you were actually working on something here that might interest me: #126

kpala picture kpala on 5 Nov 2018

I understand, cool problem by the way. So, we can improve fetchOffsets to resolve the offsets (like the consumers do), it can be optional. If you can wait a bit, we can definitely make it happen. Thanks for the explanation.

tulios picture tulios on 5 Nov 2018

Sounds like a plan, thanks a lot!

kpala picture kpala on 5 Nov 2018

@kpala issue #205 has your feature request, I'm closing this one for now.
Thanks again

tulios picture tulios on 14 Nov 2018
🎉1
Was this page helpful?
0 / 5 - 0 ratings

Related issues

plameniv picture plameniv  路  3Comments
esuarezz picture esuarezz  路  6Comments
xamgore picture xamgore  路  4Comments
marghidanu picture marghidanu  路  3Comments
pimpelsang picture pimpelsang  路  3Comments