From 4acca043af641b492f9b34c3eaa0b41830c87d5b Mon Sep 17 00:00:00 2001 From: onkar2405 Date: Thu, 6 Aug 2026 01:52:56 +0530 Subject: [PATCH] - Add opt-in `offset.validation.enabled` config (default false); when on, the replication consumer runs with auto.offset.reset=none - Catch OffsetOutOfRangeException in MirrorSourceTask.poll and classify it by the partition's log start offset: >0 means retention purged unreplicated records (DataLossException), ==0 means the topic was recreated (TopicResetException) - Log topic, partition, requested offset and log start offset before failing - Seek uncommitted partitions to the beginning in initializeConsumer, so first starts behave as they do today under auto.offset.reset=earliest - Add unit tests for both detection paths, the disabled path and the config wiring, plus integration tests using deleteRecords and topic recreation --- .../connect/mirror/DataLossException.java | 46 +++++ .../connect/mirror/MirrorSourceConfig.java | 32 +++ .../connect/mirror/MirrorSourceTask.java | 123 +++++++++++ .../connect/mirror/TopicResetException.java | 50 +++++ .../mirror/MirrorSourceConfigTest.java | 38 ++++ .../MirrorSourceTaskOffsetValidationTest.java | 193 ++++++++++++++++++ ...ectorsIntegrationOffsetValidationTest.java | 180 ++++++++++++++++ 7 files changed, 662 insertions(+) create mode 100644 connect/mirror/src/main/java/org/apache/kafka/connect/mirror/DataLossException.java create mode 100644 connect/mirror/src/main/java/org/apache/kafka/connect/mirror/TopicResetException.java create mode 100644 connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceTaskOffsetValidationTest.java create mode 100644 connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationOffsetValidationTest.java diff --git a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/DataLossException.java b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/DataLossException.java new file mode 100644 index 0000000000000..03846320a8f6e --- /dev/null +++ b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/DataLossException.java @@ -0,0 +1,46 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.connect.mirror; + +import org.apache.kafka.connect.errors.ConnectException; + +import java.io.Serial; + +/** + * Thrown when MirrorMaker 2 determines that records which were never replicated have already been + * removed from the source cluster, i.e. the offset the replication consumer wanted to read from is + * below the log start offset of the source partition. + * + *

This is unrecoverable from the connector's point of view: the missing records cannot be + * produced to the target cluster, so the task fails fast rather than silently skipping ahead to a + * later offset and leaving an undetected gap in the replicated stream. + * + *

Only raised when {@link MirrorSourceConfig#OFFSET_VALIDATION_ENABLED} is set to {@code true}. + */ +public class DataLossException extends ConnectException { + + @Serial + private static final long serialVersionUID = 1L; + + public DataLossException(String message) { + super(message); + } + + public DataLossException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConfig.java b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConfig.java index b131af9d609e2..95e3c5f31a379 100644 --- a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConfig.java +++ b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConfig.java @@ -24,6 +24,7 @@ import java.util.Map; import java.util.stream.Collectors; +import static org.apache.kafka.clients.consumer.ConsumerConfig.AUTO_OFFSET_RESET_CONFIG; import static org.apache.kafka.common.config.ConfigDef.ValidString.in; public class MirrorSourceConfig extends MirrorConnectorConfig { @@ -98,6 +99,16 @@ public class MirrorSourceConfig extends MirrorConnectorConfig { "Partition Count * offset.lag.max = Approximate duplicated record count (Actual value can be lower or even higher depending on timing and consumer lag)"; public static final long OFFSET_LAG_MAX_DEFAULT = 100L; + public static final String OFFSET_VALIDATION_ENABLED = "offset.validation.enabled"; + private static final String OFFSET_VALIDATION_ENABLED_DOC = + "Whether MirrorSourceTask should fail fast when the offset it wants to replicate from is no " + + "longer available on the source cluster. When enabled, the replication consumer is configured " + + "with auto.offset.reset=none and the task throws a DataLossException if source records were " + + "removed by the retention policy before they could be replicated, or a TopicResetException if " + + "the source topic was deleted and recreated. When disabled (the default), MirrorMaker 2 keeps " + + "its historical behaviour of silently resuming from the earliest available offset."; + public static final boolean OFFSET_VALIDATION_ENABLED_DEFAULT = false; + public static final String HEARTBEATS_REPLICATION_ENABLED = "heartbeats.replication" + ENABLED_SUFFIX; private static final String HEARTBEATS_REPLICATION_ENABLED_DOC = "Whether to replicate the heartbeats topics even when the topic filter does not include them." + " If set to true, heartbeats topics identified by the replication policy will always be replicated, regardless of the topic filter configuration." + @@ -213,6 +224,21 @@ Duration consumerPollTimeout() { return Duration.ofMillis(getLong(CONSUMER_POLL_TIMEOUT_MILLIS)); } + boolean offsetValidationEnabled() { + return getBoolean(OFFSET_VALIDATION_ENABLED); + } + + @Override + Map sourceConsumerConfig(String role) { + Map config = super.sourceConsumerConfig(role); + if (offsetValidationEnabled()) { + // Overriding the MirrorMaker 2 default of "earliest" is what lets the consumer surface an + // OffsetOutOfRangeException instead of silently rewinding to the start of the log. + config.put(AUTO_OFFSET_RESET_CONFIG, "none"); + } + return config; + } + boolean emitOffsetSyncsEnabled() { return getBoolean(EMIT_OFFSET_SYNCS_ENABLED); } @@ -320,6 +346,12 @@ private static ConfigDef defineSourceConfig(ConfigDef baseConfig) { OFFSET_LAG_MAX_DEFAULT, ConfigDef.Importance.LOW, OFFSET_LAG_MAX_DOC) + .define( + OFFSET_VALIDATION_ENABLED, + ConfigDef.Type.BOOLEAN, + OFFSET_VALIDATION_ENABLED_DEFAULT, + ConfigDef.Importance.MEDIUM, + OFFSET_VALIDATION_ENABLED_DOC) .define( OFFSET_SYNCS_TOPIC_LOCATION, ConfigDef.Type.STRING, diff --git a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceTask.java b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceTask.java index cd6b9b01eedf3..470866ef542e4 100644 --- a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceTask.java +++ b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceTask.java @@ -19,6 +19,7 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.consumer.ConsumerRecords; import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.consumer.OffsetOutOfRangeException; import org.apache.kafka.clients.producer.RecordMetadata; import org.apache.kafka.common.KafkaException; import org.apache.kafka.common.TopicPartition; @@ -36,9 +37,12 @@ import java.time.Duration; import java.util.ArrayList; +import java.util.Comparator; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.TreeMap; import java.util.concurrent.Semaphore; import java.util.stream.Collectors; @@ -59,6 +63,7 @@ public class MirrorSourceTask extends SourceTask { private boolean stopping = false; private Semaphore consumerAccess; private OffsetSyncWriter offsetSyncWriter; + private boolean offsetValidationEnabled; public MirrorSourceTask() {} @@ -66,12 +71,22 @@ public MirrorSourceTask() {} MirrorSourceTask(KafkaConsumer consumer, MirrorSourceLegacyMetrics metrics, String sourceClusterAlias, ReplicationPolicy replicationPolicy, OffsetSyncWriter offsetSyncWriter) { + this(consumer, metrics, sourceClusterAlias, replicationPolicy, offsetSyncWriter, + MirrorSourceConfig.OFFSET_VALIDATION_ENABLED_DEFAULT); + } + + // for testing + MirrorSourceTask(KafkaConsumer consumer, MirrorSourceLegacyMetrics metrics, String sourceClusterAlias, + ReplicationPolicy replicationPolicy, + OffsetSyncWriter offsetSyncWriter, + boolean offsetValidationEnabled) { this.consumer = consumer; this.legacyMetrics = metrics; this.sourceClusterAlias = sourceClusterAlias; this.replicationPolicy = replicationPolicy; consumerAccess = new Semaphore(1); this.offsetSyncWriter = offsetSyncWriter; + this.offsetValidationEnabled = offsetValidationEnabled; } @Override @@ -84,6 +99,7 @@ public void start(Map props) { metrics = metricNamesFormats.contains(METRIC_NAMES_NEW) ? config.metrics(context.pluginMetrics()) : null; pollTimeout = config.consumerPollTimeout(); replicationPolicy = config.replicationPolicy(); + offsetValidationEnabled = config.offsetValidationEnabled(); if (config.emitOffsetSyncsEnabled()) { offsetSyncWriter = new OffsetSyncWriter(config); } @@ -164,6 +180,14 @@ public List poll() { } } catch (WakeupException e) { return null; + } catch (OffsetOutOfRangeException e) { + // With auto.offset.reset=none the consumer surfaces invalid offsets instead of silently + // rewinding. Classify the cause and fail the task so the operator sees it. + if (!offsetValidationEnabled) { + log.warn("Failure during poll.", e); + return null; + } + throw classifyOffsetOutOfRange(e); } catch (KafkaException e) { log.warn("Failure during poll.", e); return null; @@ -176,6 +200,91 @@ public List poll() { } } + /** + * Works out why the replication consumer was left holding an out-of-range offset and builds the + * corresponding fail-fast exception. + * + *

For each affected partition the log start offset on the source cluster is the deciding + * signal: + *

+ * + *

Data loss takes precedence when both conditions are present in a single batch, because it + * is the condition with unrecoverable consequences for the target cluster. + * + * @param cause the exception raised by the consumer + * @return a {@link DataLossException} or a {@link TopicResetException}, never {@code null} + */ + // visible for testing + KafkaException classifyOffsetOutOfRange(OffsetOutOfRangeException cause) { + // Sort for deterministic logging and error messages. + Map requestedOffsets = new TreeMap<>( + Comparator.comparing(TopicPartition::topic).thenComparingInt(TopicPartition::partition)); + requestedOffsets.putAll(cause.offsetOutOfRangePartitions()); + + Map logStartOffsets = beginningOffsets(requestedOffsets.keySet()); + + Map dataLoss = new LinkedHashMap<>(); + Map topicReset = new LinkedHashMap<>(); + + requestedOffsets.forEach((topicPartition, requestedOffset) -> { + Long logStartOffset = logStartOffsets.get(topicPartition); + if (logStartOffset != null && logStartOffset > 0L) { + log.error("Detected data loss on {}-{}: MirrorMaker 2 requested offset {} but the " + + "earliest available offset on the source cluster is {}. {} record(s) were " + + "removed by the retention policy before they could be replicated.", + topicPartition.topic(), topicPartition.partition(), requestedOffset, + logStartOffset, logStartOffset - requestedOffset); + dataLoss.put(topicPartition, requestedOffset); + } else { + log.error("Detected a topic reset on {}-{}: MirrorMaker 2 requested offset {} but the " + + "log now starts at offset 0. The source topic was most likely deleted " + + "and recreated.", + topicPartition.topic(), topicPartition.partition(), requestedOffset); + topicReset.put(topicPartition, requestedOffset); + } + }); + + if (!dataLoss.isEmpty()) { + return new DataLossException("MirrorMaker 2 cannot replicate " + describe(dataLoss) + + " from cluster '" + sourceClusterAlias + "': the requested offsets are no longer " + + "available because the source records were removed by the retention policy. " + + "Failing the task to avoid silently skipping the missing records. Reset the " + + "connector offsets to resume replication and accept the gap.", cause); + } + + return new TopicResetException("MirrorMaker 2 cannot replicate " + describe(topicReset) + + " from cluster '" + sourceClusterAlias + "': the source topic appears to have been " + + "deleted and recreated, so the tracked offsets are no longer valid. Failing the task " + + "to avoid replicating the new topic on top of the previously mirrored data. Reset the " + + "connector offsets to resume replication.", cause); + } + + /** + * Looks up the log start offset for each partition. Any failure here is non-fatal: we fall back + * to an empty result, which classifies the failure as a topic reset and still fails the task. + */ + private Map beginningOffsets(Set topicPartitions) { + try { + return consumer.beginningOffsets(topicPartitions); + } catch (KafkaException e) { + log.warn("Unable to look up the earliest offsets for {} while classifying an " + + "out-of-range offset.", topicPartitions, e); + return Map.of(); + } + } + + private static String describe(Map offsets) { + return offsets.entrySet().stream() + .map(e -> e.getKey().topic() + "-" + e.getKey().partition() + " at offset " + e.getValue()) + .collect(Collectors.joining(", ")); + } + @Override public void commitRecord(SourceRecord record, RecordMetadata metadata) { if (stopping) { @@ -227,6 +336,11 @@ void initializeConsumer(Set taskTopicPartitions) { log.info("Starting with {} previously uncommitted partitions.", topicPartitionOffsets.values().stream() .filter(this::isUncommitted).count()); + Set uncommittedPartitions = topicPartitionOffsets.entrySet().stream() + .filter(entry -> isUncommitted(entry.getValue())) + .map(Map.Entry::getKey) + .collect(Collectors.toSet()); + topicPartitionOffsets.forEach((topicPartition, offset) -> { // Do not call seek on partitions that don't have an existing offset committed. if (isUncommitted(offset)) { @@ -237,6 +351,15 @@ void initializeConsumer(Set taskTopicPartitions) { log.trace("Seeking to offset {} for topicPartition: {}", nextOffsetToCommittedOffset, topicPartition); consumer.seek(topicPartition, nextOffsetToCommittedOffset); }); + + // Offset validation requires auto.offset.reset=none, which means the consumer has no + // fallback position for partitions we have never replicated before. Seek those explicitly to + // the beginning so that a first start behaves exactly as it does with the default settings. + if (offsetValidationEnabled && !uncommittedPartitions.isEmpty()) { + log.info("Seeking to the beginning of {} partition(s) with no previously committed offset: {}.", + uncommittedPartitions.size(), uncommittedPartitions); + consumer.seekToBeginning(uncommittedPartitions); + } } // visible for testing diff --git a/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/TopicResetException.java b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/TopicResetException.java new file mode 100644 index 0000000000000..4fec8c7ab8e65 --- /dev/null +++ b/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/TopicResetException.java @@ -0,0 +1,50 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.connect.mirror; + +import org.apache.kafka.connect.errors.ConnectException; + +import java.io.Serial; + +/** + * Thrown when MirrorMaker 2 determines that a source topic-partition has been reset -- typically + * because the topic was deleted and recreated -- while the connector still holds a committed offset + * from the previous incarnation of the topic. + * + *

The tell-tale signal is an out-of-range offset on a partition whose log start offset is + * {@code 0}: the partition has been rewound to the very beginning, so the previously tracked offset + * points past the end of the log rather than before its start. + * + *

Resuming from {@code earliest} in this situation would silently re-replicate the new topic on + * top of the old data on the target cluster, so the task fails fast and leaves the decision to an + * operator. + * + *

Only raised when {@link MirrorSourceConfig#OFFSET_VALIDATION_ENABLED} is set to {@code true}. + */ +public class TopicResetException extends ConnectException { + + @Serial + private static final long serialVersionUID = 1L; + + public TopicResetException(String message) { + super(message); + } + + public TopicResetException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConfigTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConfigTest.java index c7f7f4e19a51c..bb304a5962f2a 100644 --- a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConfigTest.java +++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceConfigTest.java @@ -107,6 +107,44 @@ public void testNonMutationOfConfigDef() { ); } + @Test + public void testOffsetValidationIsDisabledByDefault() { + MirrorSourceConfig config = new MirrorSourceConfig(makeProps()); + assertFalse(config.offsetValidationEnabled(), + "offset validation should be opt-in so that upgrades do not change failure semantics"); + assertEquals("earliest", config.sourceConsumerConfig("test").get("auto.offset.reset"), + "the replication consumer should keep the historical MirrorMaker 2 default"); + } + + @Test + public void testOffsetValidationEnabledDisablesAutoOffsetReset() { + MirrorSourceConfig config = new MirrorSourceConfig( + makeProps(MirrorSourceConfig.OFFSET_VALIDATION_ENABLED, "true")); + assertTrue(config.offsetValidationEnabled()); + assertEquals("none", config.sourceConsumerConfig("test").get("auto.offset.reset"), + "auto.offset.reset must be none so the consumer surfaces out-of-range offsets"); + } + + @Test + public void testOffsetValidationTakesPrecedenceOverExplicitAutoOffsetReset() { + // Offset validation cannot work with any other reset policy, so it wins over a user-supplied + // value rather than silently producing a configuration that never detects data loss. + MirrorSourceConfig config = new MirrorSourceConfig(makeProps( + MirrorSourceConfig.OFFSET_VALIDATION_ENABLED, "true", + MirrorConnectorConfig.CONSUMER_CLIENT_PREFIX + "auto.offset.reset", "latest")); + assertEquals("none", config.sourceConsumerConfig("test").get("auto.offset.reset")); + } + + @Test + public void testOffsetValidationDoesNotAffectOtherConsumerConfigs() { + MirrorSourceConfig config = new MirrorSourceConfig(makeProps( + MirrorSourceConfig.OFFSET_VALIDATION_ENABLED, "true", + MirrorConnectorConfig.CONSUMER_CLIENT_PREFIX + "max.poll.records", "42")); + Map consumerConfig = config.sourceConsumerConfig("test"); + assertEquals("42", consumerConfig.get("max.poll.records")); + assertEquals("false", consumerConfig.get("enable.auto.commit")); + } + @Test public void testOffsetSyncsTopic() { // Invalid location diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceTaskOffsetValidationTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceTaskOffsetValidationTest.java new file mode 100644 index 0000000000000..a65335f5ceea1 --- /dev/null +++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/MirrorSourceTaskOffsetValidationTest.java @@ -0,0 +1,193 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.connect.mirror; + +import org.apache.kafka.clients.consumer.KafkaConsumer; +import org.apache.kafka.clients.consumer.OffsetOutOfRangeException; +import org.apache.kafka.common.KafkaException; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.connect.source.SourceTaskContext; +import org.apache.kafka.connect.storage.OffsetStorageReader; + +import org.junit.jupiter.api.Test; + +import java.util.Map; +import java.util.Set; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertSame; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyMap; +import static org.mockito.ArgumentMatchers.anySet; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Tests for the log-truncation ({@link DataLossException}) and topic-reset + * ({@link TopicResetException}) detection added to {@link MirrorSourceTask}. + */ +public class MirrorSourceTaskOffsetValidationTest { + + private static final String SOURCE_CLUSTER = "primary"; + private static final TopicPartition TP0 = new TopicPartition("wal-topic", 0); + private static final TopicPartition TP1 = new TopicPartition("wal-topic", 1); + + @SuppressWarnings("unchecked") + private static KafkaConsumer mockConsumer() { + return mock(KafkaConsumer.class); + } + + private static MirrorSourceTask task(KafkaConsumer consumer, boolean offsetValidationEnabled) { + return new MirrorSourceTask(consumer, mock(MirrorSourceLegacyMetrics.class), SOURCE_CLUSTER, + new DefaultReplicationPolicy(), null, offsetValidationEnabled); + } + + @Test + public void testDataLossDetectedWhenRecordsWerePurged() { + KafkaConsumer consumer = mockConsumer(); + OffsetOutOfRangeException cause = new OffsetOutOfRangeException(Map.of(TP0, 5L)); + when(consumer.poll(any())).thenThrow(cause); + // The log now starts at offset 10, so offsets 5..9 were removed by the retention policy. + when(consumer.beginningOffsets(anySet())).thenReturn(Map.of(TP0, 10L)); + + MirrorSourceTask task = task(consumer, true); + + DataLossException e = assertThrows(DataLossException.class, task::poll); + assertSame(cause, e.getCause(), "the original consumer exception should be preserved"); + assertTrue(e.getMessage().contains("wal-topic-0"), "message should name the topic-partition"); + assertTrue(e.getMessage().contains("offset 5"), "message should name the problematic offset"); + assertTrue(e.getMessage().contains(SOURCE_CLUSTER), "message should name the source cluster"); + } + + @Test + public void testTopicResetDetectedWhenLogStartsAtZero() { + KafkaConsumer consumer = mockConsumer(); + OffsetOutOfRangeException cause = new OffsetOutOfRangeException(Map.of(TP0, 500L)); + when(consumer.poll(any())).thenThrow(cause); + // The log begins at 0 again: the topic was deleted and recreated. + when(consumer.beginningOffsets(anySet())).thenReturn(Map.of(TP0, 0L)); + + MirrorSourceTask task = task(consumer, true); + + TopicResetException e = assertThrows(TopicResetException.class, task::poll); + assertSame(cause, e.getCause(), "the original consumer exception should be preserved"); + assertTrue(e.getMessage().contains("wal-topic-0"), "message should name the topic-partition"); + assertTrue(e.getMessage().contains("offset 500"), "message should name the problematic offset"); + } + + @Test + public void testDataLossTakesPrecedenceWhenBothConditionsArePresent() { + KafkaConsumer consumer = mockConsumer(); + when(consumer.poll(any())).thenThrow(new OffsetOutOfRangeException(Map.of(TP0, 5L, TP1, 500L))); + when(consumer.beginningOffsets(anySet())).thenReturn(Map.of(TP0, 10L, TP1, 0L)); + + MirrorSourceTask task = task(consumer, true); + + DataLossException e = assertThrows(DataLossException.class, task::poll); + assertTrue(e.getMessage().contains("wal-topic-0"), "the purged partition should be reported"); + } + + @Test + public void testUnavailableLogStartOffsetIsTreatedAsTopicReset() { + KafkaConsumer consumer = mockConsumer(); + when(consumer.poll(any())).thenThrow(new OffsetOutOfRangeException(Map.of(TP0, 5L))); + // Simulate the earliest-offset lookup failing; we must still fail the task rather than + // fall through to the default "log a warning and carry on" behaviour. + when(consumer.beginningOffsets(anySet())).thenThrow(new KafkaException("broker unavailable")); + + MirrorSourceTask task = task(consumer, true); + + assertThrows(TopicResetException.class, task::poll); + } + + @Test + public void testOffsetOutOfRangeIsNotFatalWhenValidationIsDisabled() { + KafkaConsumer consumer = mockConsumer(); + when(consumer.poll(any())).thenThrow(new OffsetOutOfRangeException(Map.of(TP0, 5L))); + + MirrorSourceTask task = task(consumer, false); + + // Default MM2 behaviour: log a warning and return no records. + assertNull(task.poll()); + verify(consumer, never()).beginningOffsets(anySet()); + } + + @Test + public void testConsumerSeeksToBeginningForNewPartitionsWhenValidationIsEnabled() { + KafkaConsumer consumer = mockConsumer(); + MirrorSourceTask task = task(consumer, true); + task.initialize(offsetStorageContext()); + + task.initializeConsumer(Set.of(TP0, TP1)); + + // TP0 has a committed offset of 4, so we resume from 5. + verify(consumer, times(1)).seek(TP0, 5L); + // TP1 has never been replicated: auto.offset.reset=none gives the consumer no starting + // position, so the task must seek to the beginning explicitly. + verify(consumer, times(1)).seekToBeginning(Set.of(TP1)); + } + + @Test + public void testConsumerDoesNotSeekToBeginningWhenValidationIsDisabled() { + KafkaConsumer consumer = mockConsumer(); + MirrorSourceTask task = task(consumer, false); + task.initialize(offsetStorageContext()); + + task.initializeConsumer(Set.of(TP0, TP1)); + + verify(consumer, times(1)).seek(TP0, 5L); + // Unchanged from the default behaviour: auto.offset.reset=earliest handles new partitions. + verify(consumer, never()).seekToBeginning(anySet()); + } + + @Test + public void testClassificationIsIndependentOfPartitionOrdering() { + KafkaConsumer consumer = mockConsumer(); + when(consumer.beginningOffsets(anySet())).thenReturn(Map.of(TP1, 0L, TP0, 0L)); + + MirrorSourceTask task = task(consumer, true); + KafkaException e = task.classifyOffsetOutOfRange( + new OffsetOutOfRangeException(Map.of(TP1, 9L, TP0, 7L))); + + assertEquals(TopicResetException.class, e.getClass()); + assertTrue(e.getMessage().indexOf("wal-topic-0") < e.getMessage().indexOf("wal-topic-1"), + "partitions should be reported in a deterministic order"); + } + + /** + * A task context whose offset store has a committed offset for {@link #TP0} only. + */ + private static SourceTaskContext offsetStorageContext() { + SourceTaskContext context = mock(SourceTaskContext.class); + OffsetStorageReader offsetStorageReader = mock(OffsetStorageReader.class); + when(context.offsetStorageReader()).thenReturn(offsetStorageReader); + when(offsetStorageReader.offset(anyMap())).thenAnswer(invocation -> { + Map wrappedPartition = invocation.getArgument(0); + if (Integer.valueOf(TP0.partition()).equals(wrappedPartition.get("partition"))) { + wrappedPartition.put("offset", 4L); + } + return wrappedPartition; + }); + return context; + } +} diff --git a/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationOffsetValidationTest.java b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationOffsetValidationTest.java new file mode 100644 index 0000000000000..66242a5fe8d9e --- /dev/null +++ b/connect/mirror/src/test/java/org/apache/kafka/connect/mirror/integration/MirrorConnectorsIntegrationOffsetValidationTest.java @@ -0,0 +1,180 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.kafka.connect.mirror.integration; + +import org.apache.kafka.clients.admin.Admin; +import org.apache.kafka.clients.admin.RecordsToDelete; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.common.TopicPartition; +import org.apache.kafka.connect.mirror.DataLossException; +import org.apache.kafka.connect.mirror.MirrorSourceConfig; +import org.apache.kafka.connect.mirror.MirrorSourceConnector; +import org.apache.kafka.connect.mirror.TopicResetException; +import org.apache.kafka.connect.runtime.rest.entities.ConnectorStateInfo; +import org.apache.kafka.connect.util.clusters.EmbeddedConnectCluster; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import static org.apache.kafka.test.TestUtils.waitForCondition; + +/** + * Integration tests for the fail-fast offset validation added to + * {@link org.apache.kafka.connect.mirror.MirrorSourceTask}. + * + *

Both scenarios follow the same shape: replicate a topic normally, stop the + * {@link MirrorSourceConnector}, engineer the failure on the source cluster, then restart the + * connector and assert the task fails with the expected exception rather than quietly rewinding to + * the earliest available offset. + */ +@Tag("integration") +public class MirrorConnectorsIntegrationOffsetValidationTest extends MirrorConnectorsIntegrationBaseTest { + + private static final String TOPIC = "test-topic-1"; + private static final int TASK_FAILURE_DURATION_MS = 60_000; + private static final int ADMIN_REQUEST_TIMEOUT_MS = 60_000; + + @BeforeEach + @Override + public void startClusters() throws Exception { + // One-way replication keeps the failure attribution unambiguous. + replicateBackupToPrimary = false; + Map additionalConfig = new HashMap<>(); + additionalConfig.put("topics", "test-topic-.*"); + additionalConfig.put(MirrorSourceConfig.OFFSET_VALIDATION_ENABLED, "true"); + super.startClusters(additionalConfig); + } + + /** + * The source topic's retention policy removes records that MirrorMaker 2 has not replicated yet. + * The task must fail with a {@link DataLossException} instead of skipping the gap. + */ + @Test + public void testDataLossDetectedWhenUnreplicatedRecordsArePurged() throws Exception { + produceMessages(primaryProducer, TOPIC); + waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config, PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS); + backup.kafka().consume(NUM_RECORDS_PRODUCED, RECORD_TRANSFER_DURATION_MS, + remoteTopicName(TOPIC, PRIMARY_CLUSTER_ALIAS)); + + stopMirrorMakerConnectors(backup, MirrorSourceConnector.class); + + // Rewind the connector to the start of each partition, then purge the first half of every + // partition on the source cluster. The connector now points below the log start offset, + // which is exactly the state an aggressive retention policy would leave it in. + alterMirrorMakerSourceConnectorOffsets(backup, offset -> 0L, TOPIC); + deleteRecordsBefore(); + + backup.resumeConnector(MirrorSourceConnector.class.getSimpleName()); + + assertSourceTaskFailedWith(backup, DataLossException.class.getName()); + } + + /** + * The source topic is deleted and recreated, so the tracked offsets point past the end of a log + * that now starts at zero. The task must fail with a {@link TopicResetException} instead of + * re-replicating the new topic on top of the previously mirrored data. + */ + @Test + public void testTopicResetDetectedWhenSourceTopicIsRecreated() throws Exception { + produceMessages(primaryProducer, TOPIC); + waitUntilMirrorMakerIsRunning(backup, CONNECTOR_LIST, mm2Config, PRIMARY_CLUSTER_ALIAS, BACKUP_CLUSTER_ALIAS); + backup.kafka().consume(NUM_RECORDS_PRODUCED, RECORD_TRANSFER_DURATION_MS, + remoteTopicName(TOPIC, PRIMARY_CLUSTER_ALIAS)); + + stopMirrorMakerConnectors(backup, MirrorSourceConnector.class); + + recreateSourceTopic(); + // One record per partition, so the new log ends well before the committed offset of + // NUM_RECORDS_PER_PARTITION - 1 that the connector still holds. + produceMessages(primaryProducer, IntStream.range(0, NUM_PARTITIONS) + .mapToObj(partition -> new ProducerRecord( + TOPIC, partition, null, "recreated".getBytes())) + .collect(Collectors.toList())); + + backup.resumeConnector(MirrorSourceConnector.class.getSimpleName()); + + assertSourceTaskFailedWith(backup, TopicResetException.class.getName()); + } + + /** + * Deletes every record below {@code offset} on all partitions of the source topic, moving the + * log start offset forward the same way a retention-driven segment deletion would. + */ + private void deleteRecordsBefore() throws Exception { + Map toDelete = IntStream.range(0, NUM_PARTITIONS) + .boxed() + .collect(Collectors.toMap( + partition -> new TopicPartition(TOPIC, partition), + partition -> RecordsToDelete.beforeOffset(5))); + try (Admin admin = primary.kafka().createAdminClient()) { + admin.deleteRecords(toDelete).all().get(ADMIN_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS); + } + } + + /** + * Deletes and recreates the source topic with the same partition count, waiting for each step so + * the test does not race the controller. + */ + private void recreateSourceTopic() throws Exception { + try (Admin admin = primary.kafka().createAdminClient()) { + admin.deleteTopics(Set.of(TOPIC)).all().get(ADMIN_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS); + waitForCondition( + () -> !admin.listTopics().names().get(ADMIN_REQUEST_TIMEOUT_MS, TimeUnit.MILLISECONDS).contains(TOPIC), + ADMIN_REQUEST_TIMEOUT_MS, + "Source topic " + TOPIC + " was not deleted in time"); + } + primary.kafka().createTopic(TOPIC, NUM_PARTITIONS); + waitForTopicPartitionCreated(primary, TOPIC, NUM_PARTITIONS); + } + + /** + * Waits until at least one {@link MirrorSourceConnector} task has failed, and asserts the + * recorded stack trace names the expected exception. + */ + private static void assertSourceTaskFailedWith(EmbeddedConnectCluster cluster, String expectedExceptionName) + throws InterruptedException { + String connectorName = MirrorSourceConnector.class.getSimpleName(); + AtomicReference lastObservedTrace = new AtomicReference<>(""); + + waitForCondition(() -> { + ConnectorStateInfo status = cluster.connectorStatus(connectorName); + if (status == null) { + return false; + } + List failed = status.tasks().stream() + .filter(task -> "FAILED".equals(task.state())) + .toList(); + if (failed.isEmpty()) { + return false; + } + lastObservedTrace.set(String.valueOf(failed.get(0).trace())); + return failed.stream() + .anyMatch(task -> task.trace() != null && task.trace().contains(expectedExceptionName)); + }, TASK_FAILURE_DURATION_MS, () -> "MirrorSourceConnector task did not fail with " + + expectedExceptionName + " in time. Last observed trace: " + lastObservedTrace.get()); + } +}