ankita10r opened a new pull request, #22819:
URL: https://github.com/apache/kafka/pull/22819
# PR Description
## Summary
Part of a broader effort to make MirrorMaker 2 suitable for mission-critical
replication between a primary cluster and a disaster-recovery cluster,
beyond what vanilla MM2 offers. Adds two fault-tolerance behaviors to
`MirrorSourceTask`:
1. **Fail-fast on silent data loss.** If the source cluster's retention
purges data before MM2 replicates it, MM2 currently jumps ahead
silently with no signal. This detects the gap and fails the task
loudly instead.
2. **Self-healing on topic reset.** If the source topic is deleted and
recreated, MM2 currently stalls or throws an unhandled error. This
detects the reset and automatically resubscribes from the beginning of
the new topic.
## Problem
MM2's consumer defaults to `auto.offset.reset=earliest`. If the next
offset it needs has been purged by retention, it silently resets to
whatever data is left — no exception, no log, no signal anything was
lost. A topic delete+recreate produces the same underlying symptom
(`OffsetOutOfRangeException`), so telling the two apart needs more than
just catching the exception.
## Changes
- `MirrorSourceTask.java`:
- `auto.offset.reset=none` on the replication consumer, turning the
silent-skip into a catchable `OffsetOutOfRangeException`.
- Dedicated catch for `NoOffsetForPartitionException` — a brand-new
partition's first-ever run is not data loss.
- `primeKnownTopicIdsAndDetectResets()` checks each partition's topic
UUID *before* seeking to any stored offset, so a reset is caught even
if the old numeric offset happens to still be valid in the recreated
topic.
- The current topic UUID is stamped onto each record's committed offset
(`convertRecord`) so identity survives task/container restarts, not
just in-memory state.
- `handleOffsetOutOfRange()` remains as a secondary, reactive check for
truncation or a reset that happens mid-run without a restart.
- `ReplicationFailureClassifier.java` (new): pure decision logic —
compares two topic UUIDs, returns `TOPIC_RESET` or `DATA_LOSS`. No
Connect/consumer/admin dependency, unit-tested in isolation
(`ReplicationFailureClassifierTest.java`, 5 tests). Defaults to
`DATA_LOSS` if either UUID can't be established.
- `MirrorDataLossException.java` (new): thrown on genuine data loss, fails
the Connect task with a clear, greppable cause.
## Testing
- `ReplicationFailureClassifierTest` — 5 unit tests, pure logic, no broker.
- Kafka's pre-existing `MirrorSourceTaskTest` suite passes unchanged,
including `testSeekBehaviorDuringStart` (original
`initializeConsumer(Set<TopicPartition>)` signature preserved via an
overload).
- Full manual verification against a real two-cluster Docker stack:
normal replication, truncation fail-fast, topic-reset recovery
(including a back-to-back double-reset edge case). Automated via
`run_challenge.sh`, 6/6 scenarios passing.
Full design reasoning, including three real bugs found and fixed by
testing against a live cluster, is in `docs/DESIGN.md` of my submission
repo: https://github.com/ankita10r/kafka-dr-replication
## Scope
Contained to `connect/mirror` — one modified file plus two new small
classes and their test. No changes to `MirrorUtils`,
`MirrorConnectorConfig`, or any other connector (checkpoint, heartbeat).
--
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]
