Vineet-kumar8757 opened a new pull request, #22848:
URL: https://github.com/apache/kafka/pull/22848
## Summary
MirrorMaker 2 (MM2) can silently lose data in two operational scenarios
because
the replication consumer's auto.offset.reset defaults to earliest, which
masks the underlying OffsetOutOfRangeException:
1. *Silent data loss* – a retention policy purges source records before they
are replicated; the consumer jumps forward to a later valid offset,
leaving
an undetected gap.
2. *Source topic reset* – a source topic is deleted and recreated (log end
offset back to 0); MM2 silently restarts from the earliest offset.
This PR adds opt-in fail-fast detection for both cases.
## Design rationale
- The Kafka consumer already detects an invalid current offset and raises
org.apache.kafka.clients.consumer.OffsetOutOfRangeException — but only when
auto.offset.reset=none. So the feature forces none on the replication
consumer when enabled.
- Because none has no fallback for partitions without a committed offset,
initializeConsumer() explicitly seekToBeginning() for uncommitted
partitions to preserve MM2's original first-run behavior.
- In poll(), OffsetOutOfRangeException is caught and classified by
inspecting each partition's earliest available offset:
- earliest > 0 → records were purged → *DataLossException*
- earliest == 0 → topic was deleted/recreated → *TopicResetException*
A detailed error (topic, partition, requested offset, earliest offset) is
logged before the task fails fast. A topic reset takes precedence when both
conditions are present in the same batch.
- The behavior is gated by a new connector config
detect.offset.out.of.range.enabled (default false), so existing
deployments are unaffected unless they opt in.
## Files changed
- MirrorSourceTask.java – force auto.offset.reset=none when enabled, manual
seek-to-beginning for uncommitted partitions, catch/classify
OffsetOutOfRangeException via new handleOffsetOutOfRange().
- MirrorSourceConfig.java – new detect.offset.out.of.range.enabled config +
accessor.
- DataLossException.java (new) – thrown on source record purge.
- TopicResetException.java (new) – thrown on source topic reset.
- MirrorSourceTaskTest.java – tests for data-loss detection, topic-reset
detection, precedence, and seek-to-beginning behavior.
- MirrorSourceConfigTest.java – tests for the new config default and toggle.
## Lifecycle integration
The change lives entirely within the existing MirrorSourceTask lifecycle
(start() → initializeConsumer() → poll()). Throwing a ConnectException
subclass from poll() is the standard Connect mechanism for failing a task,
so Connect stops and surfaces the task as FAILED without any new plumbing.
## Testing
- ./gradlew :connect:mirror:test --tests "*MirrorSourceTaskTest" --tests
"*MirrorSourceConfigTest"
- ./gradlew :connect:mirror:checkstyleMain :connect:mirror:checkstyleTest
spotlessCheck
## Backward compatibility
Fully backward compatible — the feature is disabled by default. When
disabled,
MM2 retains its existing auto.offset.reset=earliest behavior.
--
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]
