Fail fast in MirrorMaker 2 when source offsets become invalid

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.
+ *
+ * <p>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.
+ *
+ * <p>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);
+    }
+}

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<String, Object> sourceConsumerConfig(String role) {
+        Map<String, Object> 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,

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,19 +63,30 @@ public class MirrorSourceTask extends SourceTask {
     private boolean stopping = false;
     private Semaphore consumerAccess;
     private OffsetSyncWriter offsetSyncWriter;
+    private boolean offsetValidationEnabled;
 
     public MirrorSourceTask() {}
 
     // for testing
     MirrorSourceTask(KafkaConsumer<byte[], byte[]> consumer, MirrorSourceLegacyMetrics metrics, String sourceClusterAlias,
                      ReplicationPolicy replicationPolicy,
                      OffsetSyncWriter offsetSyncWriter) {
+        this(consumer, metrics, sourceClusterAlias, replicationPolicy, offsetSyncWriter,
+                MirrorSourceConfig.OFFSET_VALIDATION_ENABLED_DEFAULT);
+    }
+
+    // for testing
+    MirrorSourceTask(KafkaConsumer<byte[], byte[]> 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<String, String> 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<SourceRecord> 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<SourceRecord> poll() {
         }
     }
 
+    /**
+     * Works out why the replication consumer was left holding an out-of-range offset and builds the
+     * corresponding fail-fast exception.
+     *
+     * <p>For each affected partition the log start offset on the source cluster is the deciding
+     * signal:
+     * <ul>
+     *   <li>{@code logStartOffset > 0} -- records ahead of our position were removed by the
+     *       retention policy before they could be replicated, so data has been lost.</li>
+     *   <li>{@code logStartOffset == 0} -- the log begins at the very start again, meaning the topic
+     *       was deleted and recreated (or otherwise truncated to empty) and our tracked offset now
+     *       points past the end of the log.</li>
+     * </ul>
+     *
+     * <p>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<TopicPartition, Long> requestedOffsets = new TreeMap<>(
+                Comparator.comparing(TopicPartition::topic).thenComparingInt(TopicPartition::partition));
+        requestedOffsets.putAll(cause.offsetOutOfRangePartitions());
+
+        Map<TopicPartition, Long> logStartOffsets = beginningOffsets(requestedOffsets.keySet());
+
+        Map<TopicPartition, Long> dataLoss = new LinkedHashMap<>();
+        Map<TopicPartition, Long> 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<TopicPartition, Long> beginningOffsets(Set<TopicPartition> 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<TopicPartition, Long> 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<TopicPartition> taskTopicPartitions) {
         log.info("Starting with {} previously uncommitted partitions.", topicPartitionOffsets.values().stream()
                 .filter(this::isUncommitted).count());
 
+        Set<TopicPartition> 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<TopicPartition> 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 

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.
+ *
+ * <p>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.
+ *
+ * <p>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.
+ *
+ * <p>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);
+    }
+}

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<String, Object> consumerConfig = config.sourceConsumerConfig("test");
+        assertEquals("42", consumerConfig.get("max.poll.records"));
+        assertEquals("false", consumerConfig.get("enable.auto.commit"));
+    }
+
     @Test
     public void testOffsetSyncsTopic() {
         // Invalid location