Consumer cannot reseek back to offset 0 on conflict resolution

[mosyp]

2021-08-19T16:38:45 kafka_delta_ingest [INFO] - WORKER-1: pre-process message 1:10 -- id:46
2021-08-19T16:38:46 kafka_delta_ingest::instrumentation [INFO] - WORKER-1: Delta write for version 7 has completed in 414 millis
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-1: REBALANCE - Partition assignments revoked
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: REBALANCE - Received new partition assignment list [0, 1]
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-1: REBALANCE - Received new partition assignment list [2, 3]
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 1:0 -- id:4
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: Resetting state with partitions: [1, 0]
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: Seeking consumer to 1:0
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: Partition 0 has no recorded offset. Not seeking consumer.
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 1:0 -- id:4
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:0 -- id:3
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:0 -- id:3
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:1 -- id:7
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:1 -- id:7
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:2 -- id:10
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:2 -- id:10
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:3 -- id:11
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:3 -- id:11
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:4 -- id:14
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:4 -- id:14
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:5 -- id:15
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:5 -- id:15
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:6 -- id:24
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:6 -- id:24
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:7 -- id:25
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:7 -- id:25
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:8 -- id:28
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:8 -- id:28
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:9 -- id:29
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:9 -- id:29
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-1: pre-process message 2:0 -- id:0
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-1: Resetting state with partitions: [3, 2]
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-1: Seeking consumer to 3:31
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-1: Seeking consumer to 2:26
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-1: pre-process message 2:0 -- id:0
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: Conflict offset for partition 1: state=Some(0), delta=10
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: Resetting state with partitions: [1, 0]
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: Seeking consumer to 1:10
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: Partition 0 has no recorded offset. Not seeking consumer.
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:10 -- id:36
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:10 -- id:36
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:11 -- id:37
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:11 -- id:37
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:12 -- id:38
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:12 -- id:38
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:13 -- id:39
2021-08-19T16:38:46 kafka_delta_ingest [INFO] - WORKER-2: pre-process message 0:13 -- id:39

Worker 2 is assigned to partition 0,1. But worker 1 is still working meanwhile. So when worker 2 tries to write new version, it faces the conflict on partition 1. On conflict the worker 2, gets latest offsets from delta and seeks. But there's no stored offset for partition 0, so it does not seek it, however worker 2 already consumed some messages from partition 0, which result in their absence.

[mosyp]

The fix would be: If there's no stored offset for a partition in delta, we go into auto.offset.reset if it's earliest then seek to the beginning, if latest then to the end.