Always consuming a topic from the earliest offset

[buckfullingham]

I have the following use-case:

I am storing a series of state changes in a kafka topic. The idea is that a client process should be able to rebuild its state simply by reading from the earliest point in the topic and applying all the deltas. For each topic there is a single partition (as the order of the deltas is important).

I have been playing around with the rd_kafka_subscribe and rd_kafka_assign APIs in an attempt to get librdkafka to pick up from the beginning but it appears I'm missing something (perhaps a rebalance callback?). Although debug logging shows that I have set the offset correctly (RD_KAFKA_OFFSET_BEGINNING) initially, it appears that after a few steps communicating with the broker that that information is lost and the library falls back to the auto reset behaviour (logs below).

I need the following:

1. An uninterrupted stream of data from earliest offset.
2. Notification if for some reason the stream is broken (so I can reset state and go back to the beginning of the stream).

How do I do that?

%7|1614768614.357|DESP|rdkafka#consumer-1| [thrd:main]: Adding desired topic marketdata.binance.BTCGBP [0]
%7|1614768614.357|FETCHSTART|rdkafka#consumer-1| [thrd:main]: List with 1 partition(s):
%7|1614768614.357|FETCHSTART|rdkafka#consumer-1| [thrd:main]:  marketdata.binance.BTCGBP [0] offset BEGINNING
%7|1614768614.357|OFFSETFETCH|rdkafka#consumer-1| [thrd:main]: List with 1 partition(s):
%7|1614768614.357|OFFSETFETCH|rdkafka#consumer-1| [thrd:main]:  marketdata.binance.BTCGBP [0] offset BEGINNING
%7|1614768614.362|STATE|rdkafka#consumer-1| [thrd:main]: Topic marketdata.binance.BTCGBP changed state unknown -> exists
%7|1614768614.362|PARTCNT|rdkafka#consumer-1| [thrd:main]: Topic marketdata.binance.BTCGBP partition count changed from 0 to 1
%7|1614768614.362|METADATA|rdkafka#consumer-1| [thrd:main]:   Topic marketdata.binance.BTCGBP partition 0 Leader 0
%7|1614768614.362|BRKDELGT|rdkafka#consumer-1| [thrd:main]: marketdata.binance.BTCGBP [0]: delegate to broker jamess-mbp.localdomain:9092/0 (rktp 0x7fe12180ae00, term 0, ref 3)
%7|1614768614.362|BRKDELGT|rdkafka#consumer-1| [thrd:main]: marketdata.binance.BTCGBP [0]: delegating to broker jamess-mbp.localdomain:9092/0 for partition with 0 messages (0 bytes) queued
%7|1614768614.362|BRKMIGR|rdkafka#consumer-1| [thrd:main]: Migrating topic marketdata.binance.BTCGBP [0] 0x7fe12180ae00 from (none) to jamess-mbp.localdomain:9092/0 (sending PARTITION_JOIN to jamess-mbp.localdomain:9092/0)
%7|1614768614.362|TOPBRK|rdkafka#consumer-1| [thrd:jamess-mbp.localdomain:9092/0]: jamess-mbp.localdomain:9092/0: Topic marketdata.binance.BTCGBP [0]: joining broker (rktp 0x7fe12180ae00, 0 message(s) queued)
%7|1614768614.362|METADATA|rdkafka#consumer-1| [thrd:main]: localhost:9092/bootstrap: 1/1 requested topic(s) seen in metadata
%7|1614768614.362|PAUSE|rdkafka#consumer-1| [thrd:main]: Library pausing 1 partition(s)
%7|1614768614.362|BARRIER|rdkafka#consumer-1| [thrd:main]: marketdata.binance.BTCGBP [0]: rd_kafka_toppar_op_pause_resume:2393: new version barrier v2
%7|1614768614.362|PAUSE|rdkafka#consumer-1| [thrd:main]: Pause marketdata.binance.BTCGBP [0] (v2)
%7|1614768614.362|OFFSET|rdkafka#consumer-1| [thrd:main]: Topic marketdata.binance.BTCGBP [0]: stored offset -1001, committed offset -1001: not including in commit
%7|1614768614.362|DESP|rdkafka#consumer-1| [thrd:main]: Removing (un)desired topic marketdata.binance.BTCGBP [0]
%7|1614768614.362|RESUME|rdkafka#consumer-1| [thrd:main]: Library resuming 1 partition(s)
%7|1614768614.362|BARRIER|rdkafka#consumer-1| [thrd:main]: marketdata.binance.BTCGBP [0]: rd_kafka_toppar_op_pause_resume:2393: new version barrier v3
%7|1614768614.362|RESUME|rdkafka#consumer-1| [thrd:main]: Resume marketdata.binance.BTCGBP [0] (v3)
%7|1614768614.362|OP|rdkafka#consumer-1| [thrd:main]: marketdata.binance.BTCGBP [0] received op PAUSE (v2) in fetch-state none (opv1)
%7|1614768614.362|PAUSE|rdkafka#consumer-1| [thrd:main]: Pause marketdata.binance.BTCGBP [0]: at offset INVALID (state none, v2)
%7|1614768614.362|OP|rdkafka#consumer-1| [thrd:main]: marketdata.binance.BTCGBP [0] received op PAUSE (v3) in fetch-state none (opv2)
%7|1614768614.362|RESUME|rdkafka#consumer-1| [thrd:main]: Not resuming stopped marketdata.binance.BTCGBP [0]: at offset INVALID (state none, v3) 

[edenhill]

We would need to see the output from debug=cgrp,topic,fetch to see what is going on.