feat: return isr metadata

fluss-rpc/src/main/proto/FlussApi.proto

@@ -869,8 +869,10 @@ message PbBucketMetadata {
   // optional as some time the leader may not elected yet
   optional int32 leader_id = 2;
   repeated int32 replica_id = 3 [packed = true];
-  // TODO: Add isr here.
   optional int32 leader_epoch = 4;
+  // Generation of the complete leader/ISR state. Absence indicates legacy metadata.
+  optional int32 bucket_epoch = 5;
+  repeated int32 isr_id = 6 [packed = true];
 }
 
 message PbProduceLogReqForBucket {
@@ -1410,4 +1412,4 @@ message PbLiteralValue {
   optional int64 timestamp_millis_value = 11;      // Epoch millis
   optional int32 timestamp_nano_of_millis_value = 12; // Nano of millis
   optional bytes decimal_bytes = 13;   // Serialized decimal (non-compact mode)
-}
+}

fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java

@@ -324,6 +324,12 @@ public void addUpdateMetadataRequestForTabletServers(
                 Integer leaderEpoch =
                         bucketLeaderAndIsr.map(LeaderAndIsr::leaderEpoch).orElse(null);
                 Integer leader = bucketLeaderAndIsr.map(LeaderAndIsr::leader).orElse(null);
+                List<Integer> isr =
+                        bucketLeaderAndIsr.map(LeaderAndIsr::isr).orElse(Collections.emptyList());
+                int bucketEpoch =
+                        bucketLeaderAndIsr
+                                .map(LeaderAndIsr::bucketEpoch)
+                                .orElse(BucketMetadata.NO_LEADER_ISR_STATE_EPOCH);
                 if (currentPartitionId == null) {
                     Map<Integer, List<Integer>> tableAssignment =
                             coordinatorContext.getTableAssignment(currentTableId);
@@ -332,7 +338,9 @@ public void addUpdateMetadataRequestForTabletServers(
                                     tableBucket.getBucket(),
                                     leader,
                                     leaderEpoch,
-                                    tableAssignment.get(tableBucket.getBucket()));
+                                    tableAssignment.get(tableBucket.getBucket()),
+                                    isr,
+                                    bucketEpoch);
                     updateMetadataRequestBucketMap
                             .computeIfAbsent(currentTableId, k -> new ArrayList<>())
                             .add(bucketMetadata);
@@ -346,7 +354,9 @@ public void addUpdateMetadataRequestForTabletServers(
                                     tableBucket.getBucket(),
                                     leader,
                                     leaderEpoch,
-                                    partitionAssignment.get(tableBucket.getBucket()));
+                                    partitionAssignment.get(tableBucket.getBucket()),
+                                    isr,
+                                    bucketEpoch);
                     updateMetadataRequestPartitionMap
                             .computeIfAbsent(currentTableId, k -> new HashMap<>())
                             .computeIfAbsent(tableBucket.getPartitionId(), k -> new ArrayList<>())

fluss-server/src/main/java/org/apache/fluss/server/metadata/BucketMetadata.java

@@ -19,26 +19,50 @@
 
 import javax.annotation.Nullable;
 
+import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.OptionalInt;
 
 /** This entity used to describe the bucket metadata. */
 public class BucketMetadata {
+    public static final int NO_LEADER_ISR_STATE_EPOCH = -1;
+
     private final int bucketId;
     private final @Nullable Integer leaderId;
     private final @Nullable Integer leaderEpoch;
     private final List<Integer> replicas;
+    private final List<Integer> isr;
+    private final @Nullable Integer bucketEpoch;
 
+    /**
+     * Creates legacy bucket metadata without authoritative leader/ISR state.
+     *
+     * <p>The empty ISR and absent bucket epoch distinguish this metadata from an authoritative
+     * state with an empty ISR.
+     */
     public BucketMetadata(
             int bucketId,
             @Nullable Integer leaderId,
             @Nullable Integer leaderEpoch,
             List<Integer> replicas) {
+        this(bucketId, leaderId, leaderEpoch, replicas, Collections.emptyList(), null);
+    }
+
+    public BucketMetadata(
+            int bucketId,
+            @Nullable Integer leaderId,
+            @Nullable Integer leaderEpoch,
+            List<Integer> replicas,
+            List<Integer> isr,
+            @Nullable Integer bucketEpoch) {
         this.bucketId = bucketId;
         this.leaderId = leaderId;
         this.leaderEpoch = leaderEpoch;
-        this.replicas = replicas;
+        this.replicas = Collections.unmodifiableList(new ArrayList<>(replicas));
+        this.isr = Collections.unmodifiableList(new ArrayList<>(isr));
+        this.bucketEpoch = bucketEpoch;
     }
 
     public int getBucketId() {
@@ -57,6 +81,14 @@ public List<Integer> getReplicas() {
         return replicas;
     }
 
+    public List<Integer> getIsr() {
+        return isr;
+    }
+
+    public @Nullable Integer getBucketEpoch() {
+        return bucketEpoch;
+    }
+
     @Override
     public String toString() {
         return "BucketMetadata{"
@@ -68,6 +100,10 @@ public String toString() {
                 + leaderEpoch
                 + ", replicas="
                 + replicas
+                + ", isr="
+                + isr
+                + ", bucketEpoch="
+                + bucketEpoch
                 + '}';
     }
 
@@ -83,11 +119,13 @@ public boolean equals(Object o) {
         return bucketId == that.bucketId
                 && Objects.equals(leaderId, that.leaderId)
                 && Objects.equals(leaderEpoch, that.leaderEpoch)
-                && replicas.equals(that.replicas);
+                && replicas.equals(that.replicas)
+                && isr.equals(that.isr)
+                && Objects.equals(bucketEpoch, that.bucketEpoch);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(bucketId, leaderId, leaderEpoch, replicas);
+        return Objects.hash(bucketId, leaderId, leaderEpoch, replicas, isr, bucketEpoch);
     }
 }

fluss-server/src/main/java/org/apache/fluss/server/metadata/CoordinatorMetadataProvider.java

@@ -17,6 +17,7 @@
 
 package org.apache.fluss.server.metadata;
 
+import org.apache.fluss.annotation.VisibleForTesting;
 import org.apache.fluss.metadata.PhysicalTablePath;
 import org.apache.fluss.metadata.TableBucket;
 import org.apache.fluss.metadata.TableInfo;
@@ -30,6 +31,7 @@
 import javax.annotation.Nullable;
 
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -115,17 +117,18 @@ public Optional<PhysicalTablePath> getPhysicalTablePathFromCache(long partitionI
      * Constructs bucket metadata list from coordinator context information.
      *
      * <p>This method builds a complete list of bucket metadata by combining assignment information
-     * from the provided table assignment map with leader and epoch data from the coordinator
-     * context. Each bucket's metadata includes its ID, current leader server, leader epoch, and
-     * replica assignments.
+     * from the provided table assignment map with leader and ISR data from the coordinator context.
+     * Each bucket's metadata includes its ID, current leader server, leader epoch, ISR, bucket
+     * epoch, and replica assignments.
      *
      * @param ctx the coordinator context containing leader and epoch information
      * @param tableId the table identifier
      * @param partitionId the partition identifier, null for non-partitioned tables
      * @param tableAssignment the assignment map from bucket ID to list of replica server IDs
      * @return a list of bucket metadata objects containing complete bucket information
      */
-    private static List<BucketMetadata> getBucketMetadataFromContext(
+    @VisibleForTesting
+    static List<BucketMetadata> getBucketMetadataFromContext(
             CoordinatorContext ctx,
             long tableId,
             @Nullable Long partitionId,
@@ -135,13 +138,27 @@ private static List<BucketMetadata> getBucketMetadataFromContext(
                 (bucketId, serverIds) -> {
                     TableBucket tableBucket = new TableBucket(tableId, partitionId, bucketId);
                     Optional<LeaderAndIsr> optLeaderAndIsr = ctx.getBucketLeaderAndIsr(tableBucket);
-                    Integer leader = optLeaderAndIsr.map(LeaderAndIsr::leader).orElse(null);
-                    BucketMetadata bucketMetadata =
-                            new BucketMetadata(
-                                    bucketId,
-                                    leader,
-                                    ctx.getBucketLeaderEpoch(tableBucket),
-                                    serverIds);
+                    BucketMetadata bucketMetadata;
+                    if (optLeaderAndIsr.isPresent()) {
+                        LeaderAndIsr leaderAndIsr = optLeaderAndIsr.get();
+                        bucketMetadata =
+                                new BucketMetadata(
+                                        bucketId,
+                                        leaderAndIsr.leader(),
+                                        leaderAndIsr.leaderEpoch(),
+                                        serverIds,
+                                        leaderAndIsr.isr(),
+                                        leaderAndIsr.bucketEpoch());
+                    } else {
+                        bucketMetadata =
+                                new BucketMetadata(
+                                        bucketId,
+                                        null,
+                                        null,
+                                        serverIds,
+                                        Collections.emptyList(),
+                                        BucketMetadata.NO_LEADER_ISR_STATE_EPOCH);
+                    }
                     bucketMetadataList.add(bucketMetadata);
                 });
         return bucketMetadataList;

fluss-server/src/main/java/org/apache/fluss/server/utils/ServerRpcMessageUtils.java

@@ -648,6 +648,15 @@ private static List<PbBucketMetadata> toPbBucketMetadata(
                 pbBucketMetadata.addReplicaId(replica);
             }
 
+            Integer bucketEpoch = bucketMetadata.getBucketEpoch();
+            if (bucketEpoch != null) {
+                pbBucketMetadata.setBucketEpoch(bucketEpoch);
+            }
+
+            for (Integer isr : bucketMetadata.getIsr()) {
+                pbBucketMetadata.addIsrId(isr);
+            }
+
             pbBucketMetadataList.add(pbBucketMetadata);
         }
         return pbBucketMetadataList;
@@ -688,7 +697,9 @@ private static BucketMetadata toBucketMetadata(PbBucketMetadata pbBucketMetadata
                 pbBucketMetadata.hasLeaderEpoch() ? pbBucketMetadata.getLeaderEpoch() : null,
                 Arrays.stream(pbBucketMetadata.getReplicaIds())
                         .boxed()
-                        .collect(Collectors.toList()));
+                        .collect(Collectors.toList()),
+                Arrays.stream(pbBucketMetadata.getIsrIds()).boxed().collect(Collectors.toList()),
+                pbBucketMetadata.hasBucketEpoch() ? pbBucketMetadata.getBucketEpoch() : null);
     }
 
     private static PartitionMetadata toPartitionMetadata(PbPartitionMetadata pbPartitionMetadata) {

fluss-server/src/main/java/org/apache/fluss/server/zk/ZooKeeperClient.java

@@ -1780,8 +1780,13 @@ private BucketMetadata createBucketMetadata(
         LeaderAndIsr leaderAndIsr = leaderAndIsrs.get(bucket);
         Integer leader = leaderAndIsr != null ? leaderAndIsr.leader() : null;
         Integer leaderEpoch = leaderAndIsr != null ? leaderAndIsr.leaderEpoch() : null;
+        List<Integer> isr = leaderAndIsr != null ? leaderAndIsr.isr() : Collections.emptyList();
+        int bucketEpoch =
+                leaderAndIsr != null
+                        ? leaderAndIsr.bucketEpoch()
+                        : BucketMetadata.NO_LEADER_ISR_STATE_EPOCH;
         List<Integer> replicas = assignment.getBucketAssignments().get(bucketId).getReplicas();
-        return new BucketMetadata(bucketId, leader, leaderEpoch, replicas);
+        return new BucketMetadata(bucketId, leader, leaderEpoch, replicas, isr, bucketEpoch);
     }
 
     /** Close the underlying ZooKeeperClient. */

fluss-server/src/test/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessorTest.java

@@ -1532,7 +1532,12 @@ void testSchemaChange() throws Exception {
                                         tableInfo,
                                         Collections.singletonList(
                                                 new BucketMetadata(
-                                                        0, replicas.get(0), 0, replicas)))));
+                                                        0,
+                                                        replicas.get(0),
+                                                        0,
+                                                        replicas,
+                                                        replicas,
+                                                        0)))));
 
         // alter table column.
         alterTable(
@@ -1591,7 +1596,12 @@ void testTableRegistrationChange() throws Exception {
                                         tableInfo,
                                         Collections.singletonList(
                                                 new BucketMetadata(
-                                                        0, replicas.get(0), 0, replicas)))));
+                                                        0,
+                                                        replicas.get(0),
+                                                        0,
+                                                        replicas,
+                                                        replicas,
+                                                        0)))));
 
         // alter table properties (custom property)
         TablePropertyChanges.Builder builder = TablePropertyChanges.builder();

fluss-server/src/test/java/org/apache/fluss/server/metadata/BucketMetadataTest.java

@@ -0,0 +1,83 @@
+/*
+ * 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.server.metadata;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class BucketMetadataTest {
+
+    @Test
+    void testDefensivelyCopiesAndExposesImmutableReplicaAndIsrLists() {
+        List<Integer> replicas = new ArrayList<>(Arrays.asList(1, 2, 3));
+        List<Integer> isr = new ArrayList<>(Arrays.asList(1, 2));
+        BucketMetadata metadata = new BucketMetadata(0, 1, 2, replicas, isr, 3);
+
+        replicas.add(4);
+        isr.add(3);
+
+        assertThat(metadata.getReplicas()).containsExactly(1, 2, 3);
+        assertThat(metadata.getIsr()).containsExactly(1, 2);
+        assertThatThrownBy(() -> metadata.getReplicas().add(4))
+                .isInstanceOf(UnsupportedOperationException.class);
+        assertThatThrownBy(() -> metadata.getIsr().add(3))
+                .isInstanceOf(UnsupportedOperationException.class);
+    }
+
+    @Test
+    void testEqualityHashCodeAndStringIncludeAuthoritativeState() {
+        BucketMetadata metadata =
+                new BucketMetadata(0, 1, 2, Arrays.asList(1, 2, 3), Arrays.asList(1, 2), 3);
+        BucketMetadata sameMetadata =
+                new BucketMetadata(0, 1, 2, Arrays.asList(1, 2, 3), Arrays.asList(1, 2), 3);
+        BucketMetadata differentIsr =
+                new BucketMetadata(
+                        0, 1, 2, Arrays.asList(1, 2, 3), Collections.singletonList(1), 3);
+        BucketMetadata differentEpoch =
+                new BucketMetadata(0, 1, 2, Arrays.asList(1, 2, 3), Arrays.asList(1, 2), 4);
+
+        assertThat(metadata).isEqualTo(sameMetadata).hasSameHashCodeAs(sameMetadata);
+        assertThat(metadata).isNotEqualTo(differentIsr).isNotEqualTo(differentEpoch);
+        assertThat(metadata.toString()).contains("isr=[1, 2]", "bucketEpoch=3");
+    }
+
+    @Test
+    void testAuthoritativeEmptyIsrDiffersFromLegacyUnknownIsr() {
+        BucketMetadata authoritative =
+                new BucketMetadata(
+                        0,
+                        null,
+                        null,
+                        Arrays.asList(1, 2, 3),
+                        Collections.emptyList(),
+                        BucketMetadata.NO_LEADER_ISR_STATE_EPOCH);
+        BucketMetadata legacy = new BucketMetadata(0, null, null, Arrays.asList(1, 2, 3));
+
+        assertThat(authoritative.getBucketEpoch())
+                .isEqualTo(BucketMetadata.NO_LEADER_ISR_STATE_EPOCH);
+        assertThat(legacy.getBucketEpoch()).isNull();
+        assertThat(authoritative).isNotEqualTo(legacy);
+    }
+}