sapna2 opened a new pull request, #23252: URL: https://github.com/apache/kafka/pull/23252
The problem MM2's source consumer runs with auto.offset.reset=earliest. If a previously replicated offset stops being valid — retention purged the records before MM2 got to them, or the source topic got deleted and recreated — the consumer just jumps to a new valid offset and MM2 logs a warning and moves on. For a DR pipeline that's a real problem: the replicated log now has a gap nobody knows about until it actually matters. What I changed Added DataLossException and TopicResetException (both ConnectException, so they plug into Connect's existing task-failure handling without any extra wiring). Added a fail.fast.on.offset.reset config (default false, so existing deployments see no behavior change). When enabled, the source consumer runs with auto.offset.reset=none instead of earliest, so a stale offset surfaces as OffsetOutOfRangeException rather than being silently swallowed. Added OffsetResetClassifier, which turns that exception into the right specific failure: if the partition's current earliest offset is 0, the topic was deleted and recreated → TopicResetException. If it's > 0, retention advanced past what we'd replicated → DataLossException. Both get logged with topic/partition/offset detail before the task stops. initializeConsumer needed one extra branch: with auto.offset.reset=none, a brand-new partition with no stored offset would throw immediately on first run, so we seek it to the beginning ourselves in that case. I pulled the classification logic into its own class rather than putting it directly in MirrorSourceTask.poll() — it made it possible to unit test the actual decision (reset vs. data loss vs. multi-partition) without needing to construct a full task, and it keeps poll() itself to a one-line change. Testing OffsetResetClassifierTest covers the reset case, the data-loss case (including the purged-record count in the message), multiple partitions failing in the same batch, and a defensive default when a partition is missing from beginningOffsets(). Existing MirrorSourceTaskTest / MirrorConnectorConfigTest should be re-run to confirm the default (fail.fast.on.offset.reset=false) path is unaffected. Notes Used AI assistance to work through the Kafka Connect APIs (OffsetOutOfRangeException, beginningOffsets) and draft tests quickly; the design decisions and final code are mine. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
