[test] Fix thread-safe problems in CoordinatorEventProcessorTest

Author: wuchong

fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorContext.java

@@ -402,6 +402,16 @@ protected Map<Integer, List<Integer>> getTableAssignment(long tableId) {
         return tableAssignments.getOrDefault(tableId, Collections.emptyMap());
     }
 
+    @VisibleForTesting
+    protected int replicaCounts(long tableId) {
+        return getTableAssignment(tableId).values().stream().mapToInt(List::size).sum();
+    }
+
+    @VisibleForTesting
+    protected int replicaCounts(TablePartition tablePartition) {
+        return getPartitionAssignment(tablePartition).values().stream().mapToInt(List::size).sum();
+    }
+
     @VisibleForTesting
     protected Map<Integer, List<Integer>> getPartitionAssignment(TablePartition tablePartition) {
         return partitionAssignments.getOrDefault(tablePartition, Collections.emptyMap());

fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/CoordinatorEventProcessor.java

@@ -38,6 +38,7 @@
 import com.alibaba.fluss.rpc.messages.CommitRemoteLogManifestResponse;
 import com.alibaba.fluss.rpc.messages.PbCommitLakeTableSnapshotRespForTable;
 import com.alibaba.fluss.rpc.protocol.ApiError;
+import com.alibaba.fluss.server.coordinator.event.AccessContextEvent;
 import com.alibaba.fluss.server.coordinator.event.AdjustIsrReceivedEvent;
 import com.alibaba.fluss.server.coordinator.event.CommitKvSnapshotEvent;
 import com.alibaba.fluss.server.coordinator.event.CommitLakeTableSnapshotEvent;
@@ -418,11 +419,6 @@ private void loadAssignment(
         }
     }
 
-    @VisibleForTesting
-    protected CoordinatorContext getCoordinatorContext() {
-        return coordinatorContext;
-    }
-
     private void onShutdown() {
         // first shutdown table manager
         tableManager.shutdown();
@@ -485,6 +481,11 @@ public void process(CoordinatorEvent event) {
                 completeFromCallable(
                         commitLakeTableSnapshotEvent.getRespCallback(),
                         () -> tryProcessCommitLakeTableSnapshot(commitLakeTableSnapshotEvent));
+            } else if (event instanceof AccessContextEvent) {
+                AccessContextEvent<?> accessContextEvent = (AccessContextEvent<?>) event;
+                processAccessContext(accessContextEvent);
+            } else {
+                LOG.warn("Unknown event type: {}", event.getClass().getName());
             }
         } finally {
             updateMetrics();
@@ -935,6 +936,15 @@ private CommitRemoteLogManifestResponse tryProcessCommitRemoteLogManifest(
         return response;
     }
 
+    private <T> void processAccessContext(AccessContextEvent<T> event) {
+        try {
+            T result = event.getAccessFunction().apply(coordinatorContext);
+            event.getResultFuture().complete(result);
+        } catch (Throwable t) {
+            event.getResultFuture().completeExceptionally(t);
+        }
+    }
+
     private CommitLakeTableSnapshotResponse tryProcessCommitLakeTableSnapshot(
             CommitLakeTableSnapshotEvent commitLakeTableSnapshotEvent) {
         CommitLakeTableSnapshotData commitLakeTableSnapshotData =

fluss-server/src/main/java/com/alibaba/fluss/server/coordinator/event/AccessContextEvent.java

@@ -0,0 +1,49 @@
+/*
+ *  Copyright (c) 2025 Alibaba Group Holding Ltd.
+ *
+ *  Licensed 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 com.alibaba.fluss.server.coordinator.event;
+
+import com.alibaba.fluss.server.coordinator.CoordinatorContext;
+
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+
+/**
+ * An event designed to safely access the {@link CoordinatorContext}. This is intended solely for
+ * testing purposes. Since {@link CoordinatorContext} is not thread-safe, directly accessing it in
+ * tests can lead to unsafe operations. This event ensures safe access to the {@link
+ * CoordinatorContext} during testing
+ *
+ * @param <T> the type of the result of the access operation
+ */
+public class AccessContextEvent<T> implements CoordinatorEvent {
+
+    private final Function<CoordinatorContext, T> accessFunction;
+    private final CompletableFuture<T> resultFuture;
+
+    public AccessContextEvent(Function<CoordinatorContext, T> accessFunction) {
+        this.accessFunction = accessFunction;
+        this.resultFuture = new CompletableFuture<>();
+    }
+
+    public Function<CoordinatorContext, T> getAccessFunction() {
+        return accessFunction;
+    }
+
+    public CompletableFuture<T> getResultFuture() {
+        return resultFuture;
+    }
+}