vishu160196 opened a new pull request, #23117:
URL: https://github.com/apache/kafka/pull/23117
## Summary
MirrorMaker 2's default `auto.offset.reset=earliest` masks two operationally
dangerous scenarios in PR/DR replication: aggressive source-topic retention
purging records before they're replicated, and a source topic being deleted
and recreated. In both cases the source consumer silently jumps to a valid
offset and MM2 keeps running as if nothing happened, with no signal that a
gap now exists in the replicated log.
This PR adds an opt-in `offset.validation.enabled` connector config
(default `false`, fully backward compatible) that makes `MirrorSourceTask`
fail fast instead: a new `DataLossException` is thrown when earlier,
unreplicated records were purged, and a new `TopicResetException` when the
topic was deleted and recreated.
## Design Rationale
**Files/classes changed:**
- `MirrorSourceConfig`: new `offset.validation.enabled` config (default
`false`).
- `MirrorSourceTask`:
- `start()` — when enabled, overrides the consumer's `auto.offset.reset` to
`none` instead of the default `earliest`, so the underlying
`OffsetOutOfRangeException` reaches application code instead of being
handled transparently by the consumer's built-in reset policy.
- `initializeConsumer()` — with no reset policy, the consumer can no longer
resolve a starting position on its own. For partitions with no previously
committed Connect offset (first-ever startup), this now explicitly calls
`seekToBeginning(...)` to reproduce what `earliest` used to do for free.
Without this, brand-new tasks would fail immediately with
`NoOffsetForPartitionException`.
- `poll()` — new `catch (OffsetOutOfRangeException e)` ahead of the
existing
generic `KafkaException` catch (which today just logs a warning and
silently returns, the root cause of the masking behavior). When enabled,
delegates to `classifyOffsetOutOfRange(e)` and throws; otherwise
preserves
today's behavior unchanged.
- `classifyOffsetOutOfRange()` — for each affected partition, compares the
requested (now-invalid) offset against `consumer.beginningOffsets(...)`:
`0` means the topic was recreated (`TopicResetException`), `> 0` means
retention purged a prefix of the log (`DataLossException`). Logs topic,
partition, requested offset, and earliest offset at `ERROR` before
throwing. If a batch spans partitions with mixed causes, the task treats
it as the more severe data-loss case.
- `DataLossException` / `TopicResetException` (new): unchecked, extend
`ConnectException`, so throwing them from `poll()` fails the task the same
way the file's existing `catch (Throwable e) { throw e; }` fallback already
does — no new propagation plumbing needed.
**Why this strategy:** `auto.offset.reset=none` is the only setting under
which the consumer ever surfaces `OffsetOutOfRangeException` at all — under
`earliest`/`latest` the reset happens inside `FetchCollector` before the
exception type even exists, so there's no way to detect either scenario
without this consumer config change first. Everything else follows from
that: an explicit `seekToBeginning` to keep new-task startup behavior
identical, and a catch/classify step to turn the raised exception into a
specific, actionable failure.
**Integration with the MM2 lifecycle:** Entirely additive and opt-in — with
the config left at its default (`false`), `auto.offset.reset` stays
`earliest`, `initializeConsumer()` and `poll()` are byte-for-byte unchanged
in
behavior, and `OffsetOutOfRangeException` (as a `KafkaException` subtype)
continues to fall through to the existing generic catch exactly as before.
**Known gap:** this adds a new public connector config, which per this
repo's contribution guidelines generally warrants a KIP before merging
upstream; none has been filed here since this work originated as a
self-contained take-home exercise against a personal fork.
## Test Plan
- [x] Unit tests (`MirrorSourceTaskTest`): `seekToBeginning` on enable,
`DataLossException` thrown, `TopicResetException` thrown, disabled-flag
fallback preserves today's warn-and-swallow behavior, mixed-partition
classification prefers data loss.
- [x] Config test (`MirrorSourceConfigTest`): default `false`, settable to
`true`, registered in `CONNECTOR_CONFIG_DEF`, propagates to
`MirrorSourceTaskConfig`.
- [x] Integration tests (`DedicatedMirrorIntegrationTest`, embedded
clusters):
- `testTaskFailsOnDataLossFromPurgedRecords` — drains initial replication,
stops MM2, writes further records and deletes them via
`Admin#deleteRecords` before MM2 ever consumed them, restarts MM2, and
asserts the task fails with `DataLossException` in its trace.
- `testTaskFailsOnTopicReset` — drains initial replication, stops MM2,
deletes and recreates the topic, restarts MM2, and asserts the task fails
with `TopicResetException` in its trace.
- [x] `./gradlew :connect:mirror:test` — full module suite green.
- [x] `./gradlew :connect:mirror:spotlessApply
:connect:mirror:checkstyleMain :connect:mirror:checkstyleTest` — clean.
## Evaluation Criteria (from project brief)
- **Functional correctness**: fail-fast logic implemented for both
truncation and reset; all new and existing tests pass.
- **Technical excellence**: change is opt-in and additive, confined to
`connect/mirror` (~148 LOC in `main`, excluding tests), reuses the file's
existing "throw to fail the task" idiom rather than inventing new
propagation machinery, and passes this repo's checkstyle/spotless gates.
- **Test quality**: unit tests isolate the classification and consumer-seek
logic with mocks; integration tests exercise the real embedded-cluster
failure paths end-to-end (deterministically, via `deleteRecords` and
topic delete/recreate rather than waiting on real retention timers).
- **Documentation quality**: this description covers rationale, affected
files, and lifecycle integration above.
## AI Usage Documentation
Implemented with Claude Code (Claude Sonnet 5): used to map the MM2 class
structure, analyze the consumer's
`auto.offset.reset`/`OffsetOutOfRangeException`
semantics in `clients`, implement the config/task/exception changes, and
write the unit and integration tests. All design decisions (opt-in vs
default-on, the mixed-partition data-loss-wins policy) were reviewed and
confirmed interactively before implementation.
--
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]
