Node-rdkafka: Logical offsets not accepted by KafkaConsumer `assign` method

[JaapRood]

As far as I can gather librdkafka supports logical offsets to be passed to consumer.assign(...), like earliest, latest, -1, -2, etc. However, these logical offsets seem to be ignored, falling back to the offset.

node-rdkafka version: 2.2.0
librdkafka version: 0.11.1

Use case

I'm trying to consume the partition of a topic from the beginning. I'm using this topic as a changelog (compaction by key) for a local key-value store, so it can be recovered and state be rebuilt.

Work around

Instead of using the logical offset, you can query them separately and use them in the assignment. The issue with this is that as far as I can see there's a little race condition that's being created here. In the time you request the earliest offset and at the point you use it, data could have been compacted, with the offset no longer valid.

Possible solution

I'm not sure, as I'm not familiar with C++ and Node.js bindings. From what I can make up from the docs _librdkafka_ supports it, so making sure the parameter gets passed correctly would do it? Sorry for not being more helpful here 😞 .

[webmakersteve]

It's because the offsets you pass are getting casted to integers. I'll see what I'll need to do to support this :)

Thanks for the find.

[pipefail]

@webmakersteve I encountered this exact issue, and for the exact use case outlined by @JaapRood as well. I reached the same conclusion that the JS -> C++/C cast of -1, -2 etc... was the root cause of the issue as librdkafka defines those magic constants as int64_t values.

I attempted to work around this with a variety of the native-c integer type libraries for Node.js but had no luck with any of them. My best guess at a solution would be for node-rdkafka to expose the magic constants and explicitly handle the casting when they are passed in.

[pipefail]

I should also add that this issue is important to me so if you have any suggestions on where/how you'd like to approach this issue I can start on a PR @webmakersteve. Any advice is appreciated :)

[webmakersteve]

I'll be working on this issue today. Sorry for the late response, but I haven't forgotten!

[webmakersteve]

How does this work for you? https://github.com/Blizzard/node-rdkafka/commit/8659d9c33661c1ef560189976dd4b191cbdf5bec

[JaapRood]

@webmakersteve thanks so much for you efforts! From reading the code it it looks really good. I should have time this week to test it in code as well.

[pipefail]

@webmakersteve Looks great! Will also test.

[pipefail]

@webmakersteve So I gave this a quick test and while it does look like the new TopicPartition data structure is correctly translating strings like beginning to their magic constant (-2 in that case)... I'm still not seeing the behavior I would expect from librdkafka. Here is the test case, as well as the full debugging output from librdkafka.

The consumer is instantiated with default settings and has the event.log enabled.

consumer.assign([{
    topic: 'logbus-test-topic',
    partition: 0,
    offset: 'beginning'
]});
consumer.consume();

Librdkafka Output (truncated broker init + metadata + api logs):

{ severity: 7,
  fac: 'CGRPOP',
  message: '[thrd:main]: Group "logbus-test-cg-73861" received op ASSIGN (v0) in state up (join state init, v1 vs 0)' }
{ severity: 7,
  fac: 'ASSIGN',
  message: '[thrd:main]: Group "logbus-test-cg-73861": new assignment of 1 partition(s) in join state init' }
{ severity: 7,
  fac: 'TOPIC',
  message: '[thrd:main]: New local topic: logbus-test-topic' }
{ severity: 7,
  fac: 'TOPPARNEW',
  message: '[thrd:main]: NEW logbus-test-topic [-1] 0x102410150 (at rd_kafka_topic_new0:282)' }
{ severity: 7,
  fac: 'STATE',
  message: '[thrd:main]: Topic logbus-test-topic changed state unknown -> exists' }
{ severity: 7,
  fac: 'PARTCNT',
  message: '[thrd:main]: Topic logbus-test-topic partition count changed from 0 to 1' }
{ severity: 7,
  fac: 'TOPPARNEW',
  message: '[thrd:main]: NEW logbus-test-topic [0] 0x102410a40 (at rd_kafka_topic_partition_cnt_update:509)' }
{ severity: 7,
  fac: 'METADATA',
  message: '[thrd:main]:   Topic logbus-test-topic partition 0 Leader 1001' }
{ severity: 7,
  fac: 'BRKDELGT',
  message: '[thrd:main]: logbus-test-topic [0]: delegate to broker 127.0.0.1:9092/1001 (rktp 0x102410a40, term 0, ref 2, remove 0)' }
{ severity: 7,
  fac: 'BRKDELGT',
  message: '[thrd:main]: logbus-test-topic [0]: broker 127.0.0.1:9092/1001 is now leader for partition with 0 messages (0 bytes) queued' }
{ severity: 7,
  fac: 'BRKMIGR',
  message: '[thrd:main]: Migrating topic logbus-test-topic [0] 0x102410a40 from (none) to 127.0.0.1:9092/1001 (sending PARTITION_JOIN to 127.0.0.1:9092/1001)' }
{ severity: 7,
  fac: 'DESP',
  message: '[thrd:main]: Setting topic logbus-test-topic [0] partition as desired' }
{ severity: 7,
  fac: 'BARRIER',
  message: '[thrd:main]: Group "logbus-test-cg-73861": rd_kafka_cgrp_assign:2262: new version barrier v2' }
{ severity: 7,
  fac: 'ASSIGN',
  message: '[thrd:main]: Group "logbus-test-cg-73861": assigning 1 partition(s) in join state init' }
{ severity: 7,
  fac: 'CGRPJOINSTATE',
  message: '[thrd:main]: Group "logbus-test-cg-73861" changed join state init -> assigned (v2, state up)' }
{ severity: 7,
  fac: 'BARRIER',
  message: '[thrd:main]: Group "logbus-test-cg-73861": rd_kafka_cgrp_partitions_fetch_start0:1625: new version barrier v3' }
{ severity: 7,
  fac: 'FETCHSTART',
  message: '[thrd:main]: Group "logbus-test-cg-73861": starting fetchers for 1 assigned partition(s) in join-state assigned (usable_offsets=no, v3, line 2307)' }
{ severity: 7,
  fac: 'FETCHSTART',
  message: '[thrd:main]: List with 1 partition(s):' }
{ severity: 7,
  fac: 'FETCHSTART',
  message: '[thrd:main]:  logbus-test-topic [0] offset INVALID' }
{ severity: 7,
  fac: 'OFFSET',
  message: '[thrd:main]: Fetch 1 offsets with v3' }
{ severity: 7,
  fac: 'TOPBRK',
  message: '[thrd:127.0.0.1:9092/1001]: 127.0.0.1:9092/1001: Topic logbus-test-topic [0]: joining broker (rktp 0x102410a40)' }
{ severity: 7,
  fac: 'BROADCAST',
  message: '[thrd:127.0.0.1:9092/1001]: Broadcasting state change' }
{ severity: 7,
  fac: 'OFFSET',
  message: '[thrd:main]: 127.0.0.1:9092/1001: OffsetFetchRequest(v1) for 1/1 partition(s)' }
{ severity: 7,
  fac: 'SEND',
  message: '[thrd:127.0.0.1:9092/1001]: 127.0.0.1:9092/1001: Sent OffsetFetchRequest (v1, 99 bytes @ 0, CorrId 3)' }
{ severity: 7,
  fac: 'RECV',
  message: '[thrd:127.0.0.1:9092/1001]: 127.0.0.1:9092/1001: Received OffsetFetchResponse (v1, 43 bytes, CorrId 3, rtt 1.65ms)' }
{ severity: 7,
  fac: 'OFFSETFETCH',
  message: '[thrd:main]: List with 1 partition(s):' }
{ severity: 7,
  fac: 'OFFSETFETCH',
  message: '[thrd:main]:  logbus-test-topic [0] offset INVALID' }
{ severity: 7,
  fac: 'OFFSET',
  message: '[thrd:main]: Topic logbus-test-topic [0]: setting default offset INVALID' }
{ severity: 7,
  fac: 'OFFSETFETCH',
  message: '[thrd:main]: 127.0.0.1:9092/1001: OffsetFetchResponse: logbus-test-topic [0] offset -1' }
{ severity: 7,
  fac: 'OFFFETCH',
  message: '[thrd:main]: 127.0.0.1:9092/1001: OffsetFetch for 1/1 partition(s) returned Success' }
{ severity: 7,
  fac: 'BARRIER',
  message: '[thrd:main]: Group "logbus-test-cg-73861": rd_kafka_cgrp_partitions_fetch_start0:1625: new version barrier v4' }
{ severity: 7,
  fac: 'FETCHSTART',
  message: '[thrd:main]: Group "logbus-test-cg-73861": starting fetchers for 1 assigned partition(s) in join-state assigned (usable_offsets=yes, v4, line 1554)' }
{ severity: 7,
  fac: 'FETCHSTART',
  message: '[thrd:main]: List with 1 partition(s):' }
{ severity: 7,
  fac: 'FETCHSTART',
  message: '[thrd:main]:  logbus-test-topic [0] offset INVALID' }
{ severity: 7,
  fac: 'CGRPJOINSTATE',
  message: '[thrd:main]: Group "logbus-test-cg-73861" changed join state assigned -> started (v4, state up)' }
{ severity: 7,
  fac: 'BARRIER',
  message: '[thrd:main]: logbus-test-topic [0]: rd_kafka_toppar_op_fetch_start:2056: new version barrier v2' }
{ severity: 7,
  fac: 'CONSUMER',
  message: '[thrd:main]: Start consuming logbus-test-topic [0] at offset INVALID (v2)' }
{ severity: 7,
  fac: 'OP',
  message: '[thrd:main]: logbus-test-topic [0] received op FETCH_START (v2) in fetch-state none (opv1)' }
{ severity: 7,
  fac: 'FETCH',
  message: '[thrd:main]: Start fetch for logbus-test-topic [0] in state none at offset INVALID (v2)' }
{ severity: 7,
  fac: 'PARTSTATE',
  message: '[thrd:main]: Partition logbus-test-topic [0] changed fetch state none -> offset-query' }
{ severity: 7,
  fac: 'OFFSET',
  message: '[thrd:main]: logbus-test-topic [0]: offset reset (at offset INVALID) to END: no previously committed offset available: Local: No offset stored' }
{ severity: 7,
  fac: 'OFFREQ',
  message: '[thrd:main]: 127.0.0.1:9092/1001: Partition logbus-test-topic [0]: querying for logical offset END (opv 2)' }
{ severity: 7,
  fac: 'OFFSET',
  message: '[thrd:main]: 127.0.0.1:9092/1001: OffsetRequest (v0, opv 0) for 1 topic(s) and 1 partition(s)' }
{ severity: 7,
  fac: 'PARTSTATE',
  message: '[thrd:main]: Partition logbus-test-topic [0] changed fetch state offset-query -> offset-wait' }
{ severity: 7,
  fac: 'CGRPOP',
  message: '[thrd:main]: Group "logbus-test-cg-73861" received op PARTITION_JOIN in state up (join state started, v4) for logbus-test-topic [0]' }
{ severity: 7,
  fac: 'PARTADD',
  message: '[thrd:main]: Group "logbus-test-cg-73861": add logbus-test-topic [0]' }
{ severity: 7,
  fac: 'SEND',
  message: '[thrd:127.0.0.1:9092/1001]: 127.0.0.1:9092/1001: Sent OffsetRequest (v0, 93 bytes @ 0, CorrId 4)' }
{ severity: 7,
  fac: 'RECV',
  message: '[thrd:127.0.0.1:9092/1001]: 127.0.0.1:9092/1001: Received OffsetResponse (v0, 45 bytes, CorrId 4, rtt 3.02ms)' }
{ severity: 7,
  fac: 'OFFSET',
  message: '[thrd:main]: 127.0.0.1:9092/1001: Offset reply for topic logbus-test-topic [0] (v2 vs v2)' }
{ severity: 7,
  fac: 'OFFSET',
  message: '[thrd:main]: Offset END request for logbus-test-topic [0] returned offset 108 (108)' }
{ severity: 7,
  fac: 'PARTSTATE',
  message: '[thrd:main]: Partition logbus-test-topic [0] changed fetch state offset-wait -> active' }

From what I can tell librdkafka still sees the offset provided as INVALID and it's initial check also notes usable_offsets=no which I tracked down to the logic found in this function. That said, I'd imagine it would still treat a logical offset as "unusable" in that sense that by "usable" I think it just means "absolute"... but I don't think it should be coming through as INVALID, but rather would be represented as it's logical constant name.

It looks like it then defaults to doing a logical offset query for the END offset -- as this is the fallback behavior irrespective of configuration I am assuming it's either treating the logical offset as invalid still or ignoring it?

When I run this same test but swap out beginning for a known absolute offset, it behaves exactly as expected and librdkafka accepts the absolute offset and skips the offset query. I'm not entirely sure where the loss of translation is but I'll keep digging as I can find time. Hope this is helpful and if I can provide any more information please just ask and I will.

[pipefail]

@JaapRood did you get a chance to test this? I just want to make sure I didn't mess something up throwing off my results... no rush or anything.

[JaapRood]

@petermelias I haven't had the chance yet, but should do next week. Sorry to keep everyone waiting!

[pipefail]

@webmakersteve I did verify that I tested this with version 2.3.1 by the way and librdkafka 0.11.4. I also logged the internal state of the TopicPartition structure you added and it correctly resolved the string to the Node-land integer constant, eventual C++ typecast notwithstanding. Still not entirely sure why it's still invalid when it makes it to librdkafka-land.

[webmakersteve]

Invalid offset may not mean that the value you provided was an invalid option for offsets. Invalid offsets can also mean that the offset no longer exists on the partition.

When you give it 0, it is looking for offset number 0 on the partition. If that offset has TTL'd out, it will rely on the offset.reset parameter, which defaults to latest. That's why it then seeks latest.

The special number for RD_KAFKA_OFFSET_BEGINNING is -2. Can you try providing -2?

[pipefail]

@webmakersteve I think you've misunderstood me or perhaps I was not clear enough sorry.

I am getting the INVALID output above when I call assign with -2 OR beginning. When I provide a valid (in-bounds, non-cleaned) absolute offset, everything works fine. That's the difference I was trying to highlight as I am very much aware of the INVALID state applying to expired offsets.

To be very clear -- the above output (INVALID) state is what I am seeing when I specify -2 OR beginning.

Hope that clears things up, sorry if I was unclear.

[webmakersteve]

It was my fault, sorry. I read the code you provided as setting offset to 0, not partition. My mistake. Was answering so many issues today.

I think I found the issue: https://github.com/Blizzard/node-rdkafka/commit/41696e1b55892f766ec3e08e7c63f7e6d016f77b

Can you try pulling from master and see if it works?

[pipefail]

No worries! I know how it can be and I just appreciate that you're taking any time at all :), so thank you!

Just looked at the commit -- definitely looks like the culprit! I'll test later today thanks @webmakersteve

[thebigredgeek]

Was this resolved?

[stale[bot]]

This issue has been automatically marked as stale because it has not had recent activity. It will be closed if no further activity occurs.