ldhruva987 opened a new pull request, #22852:
URL: https://github.com/apache/kafka/pull/22852
## Design Rationale
### Problem
MirrorMaker 2's `MirrorSourceTask` consumes from the source cluster with
`auto.offset.reset=earliest` (the consumer default). This masks two failure
modes that matter for Primary/DR replication:
- **Silent data loss**: if the source topic's retention policy purges
records before MM2 replicates them, the next `poll()` silently jumps to the
next available offset instead of surfacing a gap.
- **Silent topic reset**: if a source topic is deleted and recreated, the
previously tracked offset is no longer valid, and MM2 silently resets to the
earliest offset of the new topic instead of surfacing the inconsistency.
Both cases currently only produce a debug/warn-level log line buried in
`poll()`'s generic `KafkaException` handling, with no way for an operator or
the Connect framework to detect or react to the condition.
### Files and classes modified
-
`connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConfig.java`
— new connector config `data.loss.and.topic.reset.detection.enabled` (boolean,
default `false`) and accessor `dataLossAndTopicResetDetectionEnabled()`.
-
`connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceTask.java`
— core detection logic:
- `start()`: when the new config is enabled, sets `auto.offset.reset=none`
on the source consumer so `OffsetOutOfRangeException` is thrown instead of
silently swallowed.
- `initializeConsumer()`: since `auto.offset.reset=none` disables the
consumer's own fallback, partitions with no committed offset are now seeked to
the beginning manually, preserving MM2's existing startup behavior.
- `poll()`: catches `OffsetOutOfRangeException` and, when detection is
enabled, delegates to the new `classifyOffsetOutOfRange()` and rethrows its
result; otherwise falls back to the existing warn-and-continue behavior.
- `classifyOffsetOutOfRange()` (new, package-visible for testing): for
each affected partition, compares the requested offset against
`consumer.beginningOffsets()`. An earliest offset greater than 0 indicates
purged records (data loss); an earliest offset of 0 indicates the topic was
recreated (topic reset). Logs a detailed error per partition (source cluster,
topic, partition, requested offset, earliest offset) before throwing.
-
`connect/mirror/src/main/java/org/apache/kafka/connect/mirror/DataLossException.java`
(new) — thrown when purged records are detected; extends `ConnectException` so
the Connect framework fails the task.
-
`connect/mirror/src/main/java/org/apache/kafka/connect/mirror/TopicResetException.java`
(new) — thrown when a topic reset is detected; same fail-fast mechanism.
- Corresponding test updates in `MirrorSourceConfigTest.java` and
`MirrorSourceTaskTest.java`.
### Why this implementation strategy
- **Detection at the consumer boundary, not by polling metadata
separately**: `OffsetOutOfRangeException` is the exact signal Kafka's consumer
already raises when a requested offset is no longer valid, whether due to
retention or topic recreation. Using it avoids adding a second, redundant
mechanism (e.g. periodically comparing offsets against topic metadata) that
could drift out of sync with what the consumer actually observes.
- **Distinguishing data loss from topic reset using `beginningOffsets()`**:
a recreated topic's log starts at offset 0, while a topic that has simply been
truncated by retention has an earliest offset greater than 0. This lets one
exception type and one code path serve both detection scenarios cleanly.
- **Opt-in via config, defaulting to `false`**: this preserves MM2's
existing default behavior (resume from earliest, log a warning) for existing
deployments, while letting operators who need strict fail-fast semantics (e.g.
mission-critical PR/DR setups) opt in explicitly.
- **Fail-fast via dedicated exception types rather than just logging**: a
warning that's easy to miss in production logs does not stop a broken
replication pipeline from continuing to run. Throwing
`DataLossException`/`TopicResetException` (subclasses of `ConnectException`)
causes the Connect framework to fail and surface the task, so the condition
can't go unnoticed.
### Integration with the existing MM2 lifecycle
- No changes to `MirrorSourceConnector` or task lifecycle methods
(`start()`/`stop()`/`commit()`/`commitRecord()`) beyond `start()` reading the
new config and conditionally adjusting the consumer config map before the
consumer is constructed — the rest of the startup sequence is untouched.
- `initializeConsumer()` already ran on task start to seek partitions to
their committed offsets; the new branch only changes behavior for the
previously-uncommitted-offset case, and only when detection is enabled, so
default-mode task startup is unaffected.
- `poll()`'s existing exception hierarchy (`WakeupException`, generic
`KafkaException`, `Throwable`) is preserved as-is; `OffsetOutOfRangeException`
is caught ahead of the generic `KafkaException` handler so it can be
special-cased without disturbing the other branches.
- When detection is disabled (the default), behavior is bit-for-bit
identical to current MM2: same consumer config, same seek behavior, same
warn-and-continue on `OffsetOutOfRangeException`.
### Testing
- `MirrorSourceConfigTest`: verifies the new config defaults to disabled and
can be enabled.
- `MirrorSourceTaskTest`:
- `testSeekToBeginningWhenDataLossAndTopicResetDetectionEnabled` —
confirms manual `seekToBeginning` is invoked for uncommitted partitions when
detection is on.
- `testClassifyOffsetOutOfRangeAsDataLossWhenEarliestOffsetIsPositive` /
`...AsTopicResetWhenEarliestOffsetIsZero` — confirm classification logic for
both scenarios.
- `testPollPropagatesDataLossExceptionWhenDetectionEnabled` — confirms
`poll()` throws when detection is enabled.
- `testPollSwallowsOffsetOutOfRangeWhenDetectionDisabled` — confirms
default behavior (warn, return `null`, no call to `beginningOffsets()`) is
unchanged when detection is disabled.
- All new and existing `connect:mirror` tests pass (`./gradlew
:connect:mirror:test`).
--
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]
