[server] Reconcile divergent replica logs using leader epochs

fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java

@@ -985,6 +985,18 @@ public class ConfigOptions {
                                     + "we would fsync after every message; if it were 5 we would fsync after every "
                                     + "five messages.");
 
+    public static final ConfigOption<Boolean> LOG_REPLICATION_LEADER_EPOCH_ENABLED =
+            key("log.replication.leader-epoch.enabled")
+                    .booleanType()
+                    .defaultValue(true)
+                    .withDescription(
+                            "Whether to validate replication log history using leader epochs. "
+                                    + "Can be changed dynamically with alterClusterConfigs. Disabled peers remain compatible "
+                                    + "through legacy replication, which cannot verify divergent histories. "
+                                    + "Disabling discards local epoch history; enabling does not reconstruct "
+                                    + "unknown history for existing records. New boundaries are established "
+                                    + "by subsequent leader epochs, not by changing this option.");
+
     public static final ConfigOption<Duration> LOG_FLUSH_OFFSET_CHECKPOINT_INTERVAL =
             key("log.flush.offset.checkpoint-interval")
                     .durationType()

fluss-common/src/main/java/org/apache/fluss/metadata/LeaderEpochOffset.java

@@ -0,0 +1,61 @@
+/*
+ * 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.fluss.metadata;
+
+import org.apache.fluss.annotation.Internal;
+
+import java.util.Objects;
+
+/** A leader epoch and its associated log boundary. */
+@Internal
+public final class LeaderEpochOffset {
+    private final int epoch;
+    private final long offset;
+
+    public LeaderEpochOffset(int epoch, long offset) {
+        this.epoch = epoch;
+        this.offset = offset;
+    }
+
+    public int epoch() {
+        return epoch;
+    }
+
+    public long offset() {
+        return offset;
+    }
+
+    @Override
+    public boolean equals(Object other) {
+        if (!(other instanceof LeaderEpochOffset)) {
+            return false;
+        }
+        LeaderEpochOffset that = (LeaderEpochOffset) other;
+        return epoch == that.epoch && offset == that.offset;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(epoch, offset);
+    }
+
+    @Override
+    public String toString() {
+        return "LeaderEpochOffset(" + epoch + ", " + offset + ")";
+    }
+}

fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogManifestJsonSerde.java

@@ -17,6 +17,7 @@
 
 package org.apache.fluss.remote;
 
+import org.apache.fluss.metadata.LeaderEpochOffset;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.shaded.jackson2.com.fasterxml.jackson.core.JsonGenerator;
@@ -52,6 +53,7 @@ public class RemoteLogManifestJsonSerde
     private static final String MAX_TIMESTAMP_FIELD = "max_timestamp";
     private static final String SEGMENT_SIZE_IN_BYTES_FIELD = "size_in_bytes";
     private static final String HIGHEST_COPIED_END_OFFSET_FIELD = "highest_copied_end_offset";
+    private static final String LEADER_EPOCHS_FIELD = "leader_epochs";
     private static final int SNAPSHOT_VERSION = 1;
 
     @Override
@@ -99,6 +101,16 @@ public void serialize(RemoteLogManifest manifest, JsonGenerator generator) throw
             generator.writeNumberField(MAX_TIMESTAMP_FIELD, remoteLogSegment.maxTimestamp());
             generator.writeNumberField(
                     SEGMENT_SIZE_IN_BYTES_FIELD, remoteLogSegment.segmentSizeInBytes());
+            if (!remoteLogSegment.leaderEpochs().isEmpty()) {
+                generator.writeArrayFieldStart(LEADER_EPOCHS_FIELD);
+                for (LeaderEpochOffset epoch : remoteLogSegment.leaderEpochs()) {
+                    generator.writeStartObject();
+                    generator.writeNumberField("epoch", epoch.epoch());
+                    generator.writeNumberField(START_OFFSET_FIELD, epoch.offset());
+                    generator.writeEndObject();
+                }
+                generator.writeEndArray();
+            }
             generator.writeEndObject();
         }
         generator.writeEndArray();
@@ -150,6 +162,17 @@ public RemoteLogManifest deserialize(JsonNode node) {
             if (logicalEndOffsetNode != null) {
                 segmentBuilder.logicalEndOffset(logicalEndOffsetNode.asLong());
             }
+            JsonNode epochNodes = entryJson.get(LEADER_EPOCHS_FIELD);
+            if (epochNodes != null) {
+                List<LeaderEpochOffset> epochs = new ArrayList<>();
+                for (JsonNode epoch : epochNodes) {
+                    epochs.add(
+                            new LeaderEpochOffset(
+                                    epoch.get("epoch").asInt(),
+                                    epoch.get(START_OFFSET_FIELD).asLong()));
+                }
+                segmentBuilder.leaderEpochs(epochs);
+            }
             snapshotEntries.add(segmentBuilder.build());
         }
 

fluss-common/src/main/java/org/apache/fluss/remote/RemoteLogSegment.java

@@ -18,11 +18,15 @@
 package org.apache.fluss.remote;
 
 import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.metadata.LeaderEpochOffset;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.TableBucket;
 
 import javax.annotation.Nullable;
 
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
 import java.util.Objects;
 import java.util.UUID;
 
@@ -58,6 +62,7 @@ public class RemoteLogSegment {
     private final long maxTimestamp;
 
     private final int segmentSizeInBytes;
+    private final List<LeaderEpochOffset> leaderEpochs;
 
     private RemoteLogSegment(
             PhysicalTablePath physicalTablePath,
@@ -68,7 +73,8 @@ private RemoteLogSegment(
             @Nullable Long logicalStartOffset,
             @Nullable Long logicalEndOffset,
             long maxTimestamp,
-            int segmentSizeInBytes) {
+            int segmentSizeInBytes,
+            List<LeaderEpochOffset> leaderEpochs) {
         this.physicalTablePath = checkNotNull(physicalTablePath);
         this.tableBucket = checkNotNull(tableBucket);
         this.remoteLogSegmentId = checkNotNull(remoteLogSegmentId);
@@ -107,6 +113,7 @@ private RemoteLogSegment(
         }
         this.maxTimestamp = maxTimestamp;
         this.segmentSizeInBytes = segmentSizeInBytes;
+        this.leaderEpochs = Collections.unmodifiableList(new ArrayList<>(leaderEpochs));
     }
 
     public PhysicalTablePath physicalTablePath() {
@@ -161,7 +168,13 @@ public RemoteLogSegment withLogicalRange(long logicalStartOffset, long logicalEn
                 logicalStartOffset,
                 logicalEndOffset,
                 maxTimestamp,
-                segmentSizeInBytes);
+                segmentSizeInBytes,
+                leaderEpochs);
+    }
+
+    /** Epoch boundaries for this physical segment; empty for legacy manifests. */
+    public List<LeaderEpochOffset> leaderEpochs() {
+        return leaderEpochs;
     }
 
     public long maxTimestamp() {
@@ -189,7 +202,8 @@ public boolean equals(Object o) {
                 && maxTimestamp == that.maxTimestamp
                 && Objects.equals(remoteLogSegmentId, that.remoteLogSegmentId)
                 && Objects.equals(physicalTablePath, that.physicalTablePath)
-                && Objects.equals(tableBucket, that.tableBucket);
+                && Objects.equals(tableBucket, that.tableBucket)
+                && leaderEpochs.equals(that.leaderEpochs);
     }
 
     @Override
@@ -203,7 +217,8 @@ public int hashCode() {
                 logicalStartOffset,
                 logicalEndOffset,
                 maxTimestamp,
-                segmentSizeInBytes);
+                segmentSizeInBytes,
+                leaderEpochs);
     }
 
     @Override
@@ -241,6 +256,7 @@ public static class Builder {
         private @Nullable Long logicalEndOffset;
         private long maxTimestamp;
         private int segmentSizeInBytes;
+        private List<LeaderEpochOffset> leaderEpochs = Collections.emptyList();
 
         public static Builder builder() {
             return new Builder();
@@ -291,6 +307,11 @@ public Builder tableBucket(TableBucket tableBucket) {
             return this;
         }
 
+        public Builder leaderEpochs(List<LeaderEpochOffset> leaderEpochs) {
+            this.leaderEpochs = leaderEpochs;
+            return this;
+        }
+
         public RemoteLogSegment build() {
             return new RemoteLogSegment(
                     physicalTablePath,
@@ -301,7 +322,8 @@ public RemoteLogSegment build() {
                     logicalStartOffset,
                     logicalEndOffset,
                     maxTimestamp,
-                    segmentSizeInBytes);
+                    segmentSizeInBytes,
+                    leaderEpochs);
         }
     }
 }

fluss-common/src/test/java/org/apache/fluss/remote/RemoteLogManifestJsonSerdeTest.java

@@ -17,14 +17,21 @@
 
 package org.apache.fluss.remote;
 
+import org.apache.fluss.metadata.LeaderEpochOffset;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TablePath;
 import org.apache.fluss.utils.json.JsonSerdeTestBase;
+import org.apache.fluss.utils.json.JsonSerdeUtils;
+
+import org.junit.jupiter.api.Test;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.UUID;
 
+import static org.assertj.core.api.Assertions.assertThat;
+
 /** Tests of {@link RemoteLogManifestJsonSerde}. */
 class RemoteLogManifestJsonSerdeTest extends JsonSerdeTestBase<RemoteLogManifest> {
     private static final PhysicalTablePath TABLE_PATH1 =
@@ -102,6 +109,32 @@ protected RemoteLogManifestJsonSerdeTest() {
         super(RemoteLogManifestJsonSerde.INSTANCE);
     }
 
+    @Test
+    void testLeaderEpochHistorySurvivesManifestRoundTrip() throws Exception {
+        RemoteLogSegment segment =
+                RemoteLogSegment.Builder.builder()
+                        .physicalTablePath(TABLE_PATH1)
+                        .tableBucket(TABLE_BUCKET1)
+                        .remoteLogSegmentId(UUID.randomUUID())
+                        .remoteLogStartOffset(10)
+                        .remoteLogEndOffset(30)
+                        .maxTimestamp(0)
+                        .segmentSizeInBytes(100)
+                        .leaderEpochs(
+                                Arrays.asList(
+                                        new LeaderEpochOffset(4, 5), new LeaderEpochOffset(7, 20)))
+                        .build();
+        RemoteLogManifest manifest =
+                new RemoteLogManifest(
+                        TABLE_PATH1, TABLE_BUCKET1, Collections.singletonList(segment));
+        byte[] json =
+                JsonSerdeUtils.writeValueAsBytes(manifest, RemoteLogManifestJsonSerde.INSTANCE);
+        assertThat(JsonSerdeUtils.readValue(json, RemoteLogManifestJsonSerde.INSTANCE))
+                .isEqualTo(manifest);
+        assertThat(segment.withLogicalRange(15, 25).leaderEpochs())
+                .containsExactly(new LeaderEpochOffset(4, 5), new LeaderEpochOffset(7, 20));
+    }
+
     @Override
     protected RemoteLogManifest[] createObjects() {
         return new RemoteLogManifest[] {MANIFEST_SNAPSHOT1, MANIFEST_SNAPSHOT2};

fluss-rpc/src/main/java/org/apache/fluss/rpc/entity/FetchLogEpochInfo.java

@@ -0,0 +1,57 @@
+/*
+ * 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.fluss.rpc.entity;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.metadata.LeaderEpochOffset;
+
+import javax.annotation.Nullable;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/** Epoch validation and log history attached to a replica fetch response. */
+@Internal
+public final class FetchLogEpochInfo {
+    private final int leaderEpoch;
+    @Nullable private final LeaderEpochOffset divergingEpoch;
+    private final List<LeaderEpochOffset> epochStarts;
+
+    public FetchLogEpochInfo(
+            int leaderEpoch,
+            @Nullable LeaderEpochOffset divergingEpoch,
+            List<LeaderEpochOffset> epochStarts) {
+        this.leaderEpoch = leaderEpoch;
+        this.divergingEpoch = divergingEpoch;
+        this.epochStarts = Collections.unmodifiableList(new ArrayList<>(epochStarts));
+    }
+
+    public int leaderEpoch() {
+        return leaderEpoch;
+    }
+
+    @Nullable
+    public LeaderEpochOffset divergingEpoch() {
+        return divergingEpoch;
+    }
+
+    public List<LeaderEpochOffset> epochStarts() {
+        return epochStarts;
+    }
+}