swuferhong commented on code in PR #3391: URL: https://github.com/apache/fluss/pull/3391#discussion_r3338196341
########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/QueueItem.java: ########## @@ -0,0 +1,87 @@ +/* + * 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.coordinator.channel; + +import org.apache.fluss.metadata.TableBucketReplica; +import org.apache.fluss.rpc.messages.ApiMessage; + +import javax.annotation.Nullable; + +import java.util.Set; +import java.util.function.BiConsumer; + +/** + * An immutable item placed on the per-tablet-server sender queue. + * + * <p>Mirrors Kafka's {@code QueueItem} (ControllerChannelManager.scala:218-219) with two + * Fluss-specific additions: {@code coordinatorEpoch} (for the §5.4.1 stale-drop check that Kafka + * does not need because controller resignation tears down the channel manager before the new epoch + * takes over) and {@code deletionReplicas} (so the response-handler path can reconstruct the + * replica set without re-deriving it from the response). + */ +public final class QueueItem { + + private final ApiKey apiKey; + private final ApiMessage request; + @Nullable private final BiConsumer<ApiMessage, Throwable> callback; + private final int coordinatorEpoch; + private final long enqueueTimeMs; + @Nullable private final Set<TableBucketReplica> deletionReplicas; + + @SuppressWarnings("unchecked") + public QueueItem( + ApiKey apiKey, + ApiMessage request, + @Nullable BiConsumer<? extends ApiMessage, ? super Throwable> callback, + int coordinatorEpoch, + long enqueueTimeMs, + @Nullable Set<TableBucketReplica> deletionReplicas) { + this.apiKey = apiKey; + this.request = request; + this.callback = (BiConsumer<ApiMessage, Throwable>) callback; + this.coordinatorEpoch = coordinatorEpoch; + this.enqueueTimeMs = enqueueTimeMs; + this.deletionReplicas = deletionReplicas; + } + + public ApiKey getApiKey() { + return apiKey; + } + + public ApiMessage getRequest() { + return request; + } + + @Nullable + public BiConsumer<ApiMessage, Throwable> getCallback() { + return callback; + } + + public int getCoordinatorEpoch() { + return coordinatorEpoch; + } + + public long getEnqueueTimeMs() { + return enqueueTimeMs; + } + + @Nullable + public Set<TableBucketReplica> getDeletionReplicas() { Review Comment: This method and this field both are not used. I think it need to be removed ########## fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java: ########## @@ -859,6 +859,25 @@ public class ConfigOptions { .withDescription("The amount of time to sleep when fetch bucket error occurs.") .withFallbackKeys("log.replica.fetch-backoff-interval"); + public static final ConfigOption<Duration> COORDINATOR_REQUEST_RETRY_BACKOFF = Review Comment: Why are the two `coordinator-related` config options placed together with the `log-related` ones? Could we move them up to the server module section instead? Also, please add documentation for these options in `configuration.md`. ########## fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java: ########## @@ -859,6 +859,25 @@ public class ConfigOptions { .withDescription("The amount of time to sleep when fetch bucket error occurs.") .withFallbackKeys("log.replica.fetch-backoff-interval"); + public static final ConfigOption<Duration> COORDINATOR_REQUEST_RETRY_BACKOFF = + key("coordinator.request-retry.backoff-interval") + .durationType() + .defaultValue(Duration.ofMillis(100)) + .withDescription( + "The backoff duration the coordinator waits before retrying a " + + "control-plane request to a tablet server after a " + + "transient RPC-layer failure. Mirrors Kafka's " Review Comment: `Mirrors Kafka's ControllerChannelManager retry backoff (hardcoded 100ms)` -> Suggest removing this part — there's no need to mention it here. ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/ApiKey.java: ########## @@ -0,0 +1,30 @@ +/* + * 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.coordinator.channel; + +/** + * Enumeration of control-plane RPCs routed through the per-tablet-server sender thread. + * + * <p>Only {@link #STOP_REPLICA} is supported in this iteration. Future iterations may add other + * coordinator → tablet-server RPCs (e.g., {@code NOTIFY_LEADER_AND_ISR}) once they are migrated to + * the same sender-thread retry layer. + */ +public enum ApiKey { Review Comment: I don't think we need to introduce this new enum. We already have `org.apache.fluss.rpc.protocol.ApiKeys` in fluss-rpc, which is the canonical enum for all wire-protocol APIs and already includes `STOP_REPLICA` — along with all the other control-plane RPCs (`NOTIFY_LEADER_AND_ISR`, `UPDATE_METADATA`, `NOTIFY_REMOTE_LOG_OFFSETS`, etc.). The new ApiKey only has a single member today, and as we migrate more control-plane RPCs onto the sender thread, its members would just duplicate the ones already in `ApiKeys`. That gives us two parallel "ApiKey" concepts to keep in sync, which seems unnecessary. There's also no dependency concern: fluss-server already depends on fluss-rpc and uses `ApiKeys `elsewhere. ########## fluss-common/src/main/java/org/apache/fluss/metrics/MetricNames.java: ########## @@ -45,6 +45,14 @@ public class MetricNames { public static final String PARTITION_COUNT = "partitionCount"; public static final String REPLICAS_TO_DELETE_COUNT = "replicasToDeleteCount"; + // for coordinator sender (per-tablet-server control request sender threads) + public static final String SENDER_TOTAL_QUEUE_SIZE = "senderTotalQueueSize"; Review Comment: `senderTotalQueueSize` -> this one can be dropped? It's derivable by aggregating senderQueueSize across tablet servers, so it's not strictly necessary. If you'd like to keep a single "global at-a-glance" metric, leaving it in is fine too — it's a judgment call. But if we're optimizing for a leaner metric set, I'd prefer to drop this one rather than the per-TS gauge, since the per-TS breakdown carries more diagnostic value (it tells you which tablet server is backing up). ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/QueueItem.java: ########## @@ -0,0 +1,87 @@ +/* + * 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.coordinator.channel; + +import org.apache.fluss.metadata.TableBucketReplica; +import org.apache.fluss.rpc.messages.ApiMessage; + +import javax.annotation.Nullable; + +import java.util.Set; +import java.util.function.BiConsumer; + +/** + * An immutable item placed on the per-tablet-server sender queue. + * + * <p>Mirrors Kafka's {@code QueueItem} (ControllerChannelManager.scala:218-219) with two Review Comment: ditto. If you want to indicate that you referenced an Apache open source project, please use a more standard comment format ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/ControlRequestSendThread.java: ########## @@ -0,0 +1,204 @@ +/* + * 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.coordinator.channel; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metrics.Counter; +import org.apache.fluss.metrics.DescriptiveStatisticsHistogram; +import org.apache.fluss.metrics.Histogram; +import org.apache.fluss.metrics.MetricNames; +import org.apache.fluss.metrics.groups.MetricGroup; +import org.apache.fluss.rpc.gateway.TabletServerGateway; +import org.apache.fluss.rpc.messages.ApiMessage; +import org.apache.fluss.rpc.messages.StopReplicaRequest; +import org.apache.fluss.utils.concurrent.ShutdownableThread; + +import java.util.Optional; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.IntSupplier; +import java.util.function.Supplier; + +/** + * Per-tablet-server sender thread that drains the control-plane request queue and retries on + * transient RPC failures. Mirrors Kafka's {@code RequestSendThread} + * (ControllerChannelManager.scala:226-300). + * + * <p>Each invocation of {@link #doWork()} takes one {@link QueueItem} from the queue. Stale items + * (whose {@code coordinatorEpoch} is less than the current epoch) are dropped immediately. + * Otherwise the item is sent to the tablet server via the gateway, retrying with a configurable + * backoff until the send succeeds or the thread is shut down. + * + * <p>The callback is invoked in its own {@code try/catch} OUTSIDE the retry loop (alignment fix F), + * so a buggy response handler cannot re-enter the send retry path and cannot reach the outer + * Scenario-E catch. + */ +public class ControlRequestSendThread extends ShutdownableThread { + + private static final int HISTOGRAM_WINDOW_SIZE = 100; Review Comment: This field is never used. ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorChannelManager.java: ########## @@ -191,6 +345,13 @@ protected Optional<TabletServerGateway> getTabletServerGateway(int targetServerI return rpcGatewayManager.getRpcGateway(targetServerId); } + @VisibleForTesting + Map<Integer, TabletServerChannelState> getChannelStates() { Review Comment: No use. pls remove it. ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/TabletServerChannelState.java: ########## @@ -0,0 +1,93 @@ +/* + * 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.coordinator.channel; + +import org.apache.fluss.cluster.ServerNode; +import org.apache.fluss.metrics.groups.MetricGroup; + +import javax.annotation.concurrent.ThreadSafe; + +import java.util.concurrent.BlockingQueue; + +/** + * Per-tablet-server channel state held by {@link + * org.apache.fluss.server.coordinator.CoordinatorChannelManager}: the queue, the sender thread, and + * the metric handles needed to deregister the per-TS metrics on remove. + * + * <p>Mirrors Kafka's {@code ControllerBrokerStateInfo} (ControllerChannelManager.scala:768-774). + * The {@code networkClient} field is intentionally absent — {@code RpcGatewayManager} owns the + * gateway exactly the way Kafka's controller owns the {@code NetworkClient}, so we do not duplicate + * it on the state. Three Fluss-specific additions surface behavior that has no Kafka analog: + * + * <ul> + * <li>{@code retryCount} — counts §5.4 retry attempts. + * <li>{@code staleDropCount} — counts §5.4.1 stale-epoch drops at dequeue. + * <li>{@code aliveGauge} — presence indicator for the per-TS sender (§8.5 Scenario E). + * </ul> + * + * <p>Each field is final and assigned once; mutable concurrent state is encapsulated in the queue, + * the thread, and the metric handles themselves. + */ +@ThreadSafe +public final class TabletServerChannelState { + + private final int tabletServerId; + private final ServerNode serverNode; + private final BlockingQueue<QueueItem> queue; + private final ControlRequestSendThread sendThread; + + /** + * The per-TS child {@link MetricGroup}. Held so {@code removeTabletServer} can {@code close()} + * it and deregister all five per-TS metrics in one shot — mirrors Kafka's two {@code + * metricsGroup.removeMetric(...)} calls at ControllerChannelManager.scala:203-204. + */ + private final MetricGroup metricGroup; + + public TabletServerChannelState( + int tabletServerId, + ServerNode serverNode, + BlockingQueue<QueueItem> queue, + ControlRequestSendThread sendThread, + MetricGroup metricGroup) { + this.tabletServerId = tabletServerId; + this.serverNode = serverNode; + this.queue = queue; + this.sendThread = sendThread; + this.metricGroup = metricGroup; + } + + public int getTabletServerId() { Review Comment: No user. remove it. ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java: ########## @@ -1116,6 +1206,32 @@ private void processNewTabletServer(NewTabletServerEvent newTabletServerEvent) { // and offline partitions to see if those tablet servers become leaders for some/all // of those tableBucketStateMachine.triggerOnlineBucketStateChange(); + + // Clear ineligible marks for tables/partitions with replicas on the reconnecting + // server and resume their deletion. + Set<TableBucketReplica> replicasOnReconnectingServer = + coordinatorContext.replicasOnTabletServer(tabletServerId); + Set<Long> tablesToResume = + replicasOnReconnectingServer.stream() + .map(r -> r.getTableBucket().getTableId()) + .filter(coordinatorContext::isTableQueuedForDeletion) + .collect(Collectors.toSet()); + Set<TablePartition> partitionsToResume = + replicasOnReconnectingServer.stream() + .filter(r -> r.getTableBucket().getPartitionId() != null) + .map( + r -> + new TablePartition( + r.getTableBucket().getTableId(), + r.getTableBucket().getPartitionId())) + .filter(coordinatorContext::isPartitionQueuedForDeletion) + .collect(Collectors.toSet()); + if (!tablesToResume.isEmpty() || !partitionsToResume.isEmpty()) { + tablesToResume.forEach(coordinatorContext::removeTableFromIneligibleForDeletion); + partitionsToResume.forEach( + coordinatorContext::removePartitionFromIneligibleForDeletion); + tableManager.resumeDeletions(); + } } private void processDeadTabletServer(DeadTabletServerEvent deadTabletServerEvent) { Review Comment: Even on the legitimate "TS went offline" path, if that TS's machine is decommissioned and never rejoins under the same id, the table stays ineligible and its deletion hangs forever — there's no timeout, alert, or manual force-clear entry point. Kafka has the same limitation (it also relies on a broker restart), so this is a known constraint rather than a new regression. - at minimum, maybe we can add an observability hook so operators can detect a stuck deletion — e.g. a metric for "time a table/partition has been ineligible for deletion," or a periodic WARN log listing tables/partitions that have been ineligible beyond some threshold. ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorChannelManager.java: ########## @@ -35,56 +39,145 @@ import org.apache.fluss.rpc.messages.StopReplicaResponse; import org.apache.fluss.rpc.messages.UpdateMetadataRequest; import org.apache.fluss.rpc.messages.UpdateMetadataResponse; +import org.apache.fluss.server.coordinator.channel.ApiKey; +import org.apache.fluss.server.coordinator.channel.ControlRequestSendThread; +import org.apache.fluss.server.coordinator.channel.QueueItem; +import org.apache.fluss.server.coordinator.channel.TabletServerChannelState; import org.apache.fluss.server.utils.RpcGatewayManager; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.annotation.concurrent.NotThreadSafe; +import javax.annotation.Nullable; +import java.util.ArrayList; import java.util.Collection; +import java.util.HashMap; +import java.util.Map; import java.util.Optional; +import java.util.Set; +import java.util.concurrent.BlockingQueue; import java.util.concurrent.CompletableFuture; +import java.util.concurrent.LinkedBlockingQueue; import java.util.function.BiConsumer; +import java.util.function.IntSupplier; import static org.apache.fluss.utils.Preconditions.checkState; /** - * Using by coordinator server. It's a manager to manage the rpc channels to tablet servers and send - * request to the servers. + * Used by the coordinator server to manage RPC channels to tablet servers and send requests. + * Mutations are guarded by {@code channelLock} so the metric reporter thread can safely read queue + * sizes. Mirrors Kafka's {@code ControllerChannelManager} which uses a {@code brokerLock} for the + * same purpose. */ -@NotThreadSafe public class CoordinatorChannelManager { private static final Logger LOG = LoggerFactory.getLogger(CoordinatorChannelManager.class); + private static final int WINDOW_SIZE = 100; + /** A manager for the rpc gateways to tablet servers. */ private final RpcGatewayManager<TabletServerGateway> rpcGatewayManager; - public CoordinatorChannelManager(RpcClient rpcClient) { + private final IntSupplier epochSupplier; + private final Configuration conf; + private final MetricGroup coordinatorMetricGroup; + + private final Object channelLock = new Object(); + private final Map<Integer, TabletServerChannelState> channelStates = new HashMap<>(); + + public CoordinatorChannelManager( + RpcClient rpcClient, + IntSupplier epochSupplier, + Configuration conf, + MetricGroup coordinatorMetricGroup) { this.rpcGatewayManager = new RpcGatewayManager<>(rpcClient, TabletServerGateway.class); + this.epochSupplier = epochSupplier; + this.conf = conf; + this.coordinatorMetricGroup = coordinatorMetricGroup; + coordinatorMetricGroup.gauge( + MetricNames.SENDER_TOTAL_QUEUE_SIZE, + () -> { + synchronized (channelLock) { + int total = 0; + for (TabletServerChannelState s : channelStates.values()) { + total += s.getQueue().size(); + } + return total; + } + }); } public void startup(Collection<ServerNode> serverNodes) { for (ServerNode serverNode : serverNodes) { - addTabletServer(serverNode); + addNewTabletServer(serverNode); + } + synchronized (channelLock) { + for (TabletServerChannelState state : channelStates.values()) { + startSendThread(state); + } } } public void close() throws Exception { rpcGatewayManager.close(); } + /** Adds a tablet server and immediately starts its sender thread (runtime addition). */ public void addTabletServer(ServerNode serverNode) { - // add new tablet server to the channel manager + addNewTabletServer(serverNode); + synchronized (channelLock) { + TabletServerChannelState state = channelStates.get(serverNode.id()); + if (state != null) { + startSendThread(state); + } + } + } + + private void addNewTabletServer(ServerNode serverNode) { checkState( serverNode.serverType().equals(ServerType.TABLET_SERVER), "The server type should be TABLET_SERVER, but was " + serverNode.serverType()); rpcGatewayManager.addServer(serverNode); + + int id = serverNode.id(); + synchronized (channelLock) { + if (channelStates.containsKey(id)) { + return; + } + BlockingQueue<QueueItem> queue = new LinkedBlockingQueue<>(); + + MetricGroup tsGroup = + coordinatorMetricGroup.addGroup("tablet-server-id", String.valueOf(id)); + tsGroup.gauge(MetricNames.SENDER_QUEUE_SIZE, queue::size); + + ControlRequestSendThread thread = + new ControlRequestSendThread( + id, + queue, + () -> rpcGatewayManager.getRpcGateway(id), + epochSupplier, + conf, + tsGroup); + channelStates.put( + id, new TabletServerChannelState(id, serverNode, queue, thread, tsGroup)); + } + } + + private void startSendThread(TabletServerChannelState state) { + ControlRequestSendThread thread = state.getSendThread(); + if (thread.getState() == Thread.State.NEW) { + thread.start(); + } } public void removeTabletServer(Integer serverId) { + synchronized (channelLock) { + TabletServerChannelState state = channelStates.remove(serverId); + teardownChannelState(serverId, state); Review Comment: Maybe this need to remove outside of the lock. ShutdownableThread.shutdown() blocks until the thread is fully joined, and we call it inside the synchronized (channelLock) block. The same lock guards three hot paths: - the SENDER_TOTAL_QUEUE_SIZE gauge lambda — read by the metric-reporter thread; - sendStopBucketReplicaRequest when it looks up the channel state - getChannelStates() So for as long as the join takes, metric reporting and control-plane sends to all other tablet servers are blocked on a single TS teardown. ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java: ########## @@ -919,43 +959,93 @@ private void processDropPartition(DropPartitionEvent dropPartitionEvent) { dropTableInfo.getTablePath(), tableId, tablePartition.getPartitionId()); } + /** + * Handles StopReplica deletion responses from the sender thread. Partitions per-bucket results + * into succeeded/failed sets, then moves replicas to the appropriate state. + * + * <p>Mirrors Kafka's {@code processTopicDeletionStopReplicaResponseReceived} + * (KafkaController.scala:1451-1471). Fluss's StopReplicaResponse has no top-level error code + * (FlussApi.proto:373-375), so the Kafka "requestError != NONE → all in error" branch does not + * apply. Each bucket's error code is evaluated individually. + * + * <p>No {@code isActive} guard is needed (unlike Kafka :1454): {@code + * coordinatorEventManager.close()} in {@link #shutdown()} joins the event thread before {@code + * coordinatorChannelManager.shutdown()}, so no event runs after resignation. + */ private void processDeleteReplicaResponseReceived( DeleteReplicaResponseReceivedEvent deleteReplicaResponseReceivedEvent) { List<DeleteReplicaResultForBucket> deleteReplicaResultForBuckets = deleteReplicaResponseReceivedEvent.getDeleteReplicaResults(); - Set<TableBucketReplica> failDeletedReplicas = new HashSet<>(); - Set<TableBucketReplica> successDeletedReplicas = new HashSet<>(); - for (DeleteReplicaResultForBucket deleteReplicaResultForBucket : - deleteReplicaResultForBuckets) { - TableBucketReplica tableBucketReplica = - deleteReplicaResultForBucket.getTableBucketReplica(); - if (deleteReplicaResultForBucket.succeeded()) { - successDeletedReplicas.add(tableBucketReplica); + Set<TableBucketReplica> succeeded = new HashSet<>(); + Set<TableBucketReplica> failed = new HashSet<>(); + for (DeleteReplicaResultForBucket result : deleteReplicaResultForBuckets) { + if (result.succeeded()) { + succeeded.add(result.getTableBucketReplica()); } else { - failDeletedReplicas.add(tableBucketReplica); + failed.add(result.getTableBucketReplica()); } } - // clear the fail deleted number for the success deleted replicas - coordinatorContext.clearFailDeleteNumbers(successDeletedReplicas); - // pick up the replicas to retry delete and replicas that considered as success delete - Tuple2<Set<TableBucketReplica>, Set<TableBucketReplica>> - retryDeleteAndSuccessDeleteReplicas = - coordinatorContext.retryDeleteAndSuccessDeleteReplicas(failDeletedReplicas); + if (!failed.isEmpty()) { + failReplicaDeletion(failed); Review Comment: `failReplicaDeletion` marks the owning table/partition ineligible unconditionally, without checking whether the owning `TabletServer` is still alive. But the only way to reach `failReplicaDeletion` is a response whose bucket result is an error — i.e. the TS actually replied and is alive. (Pure network failures are retried indefinitely by the per-TS sender thread and never surface here. Meanwhile, the only code path that clears the table/partition ineligible flag is `processNewTabletServer`, which fires on a TS reconnect. A live TS never reconnects, so the flag is never cleared. That produces a permanent hang: ``` stopReplica returns error (TS alive) → failReplicaDeletion: replica → Ineligible, table marked ineligible → resumeTableDeletions Step 2 (no Started + has Ineligible): Ineligible → Offline → resumeTableDeletions Step 3 isEligibleForDeletion = queued && noStarted && !ineligible → table still in ineligible set → false → not deleted → replica parks at OfflineReplica; table stays queued + ineligible forever → next round: no Ineligible left (Step 2 won't fire), table flag still set (Step 3 still blocks) → TS stays alive, never reconnects → flag never cleared → permanent hang ``` This isn't an extreme corner case — it's the ordinary "TS alive but deletion failed" failure path. Suggested fix: - In failReplicaDeletion, do not mark the table ineligible for a still-alive TS — just retry with backoff; only mark ineligible when the TS is confirmed offline. ########## fluss-common/src/main/java/org/apache/fluss/config/ConfigOptions.java: ########## @@ -859,6 +859,25 @@ public class ConfigOptions { .withDescription("The amount of time to sleep when fetch bucket error occurs.") .withFallbackKeys("log.replica.fetch-backoff-interval"); + public static final ConfigOption<Duration> COORDINATOR_REQUEST_RETRY_BACKOFF = + key("coordinator.request-retry.backoff-interval") Review Comment: The two new options use inconsistent key structures: `coordinator.request-retry.backoff-interval` — uses a request-retry segment `coordinator.request.timeout` — uses a request segment These are both about the same thing (control-plane requests from the coordinator), so they should share a common coordinator.request. prefix. Right now request-retry introduces a separate sub-namespace that doesn't line up with request.timeout.Suggested rename to keep them under one prefix: `coordinator.request-retry.backoff-interval`→ `coordinator.request.retry-backoff` `coordinator.request.timeout` → keep as is ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java: ########## @@ -919,43 +959,93 @@ private void processDropPartition(DropPartitionEvent dropPartitionEvent) { dropTableInfo.getTablePath(), tableId, tablePartition.getPartitionId()); } + /** + * Handles StopReplica deletion responses from the sender thread. Partitions per-bucket results + * into succeeded/failed sets, then moves replicas to the appropriate state. + * + * <p>Mirrors Kafka's {@code processTopicDeletionStopReplicaResponseReceived} Review Comment: Please don't add so many reference comments in the code. Just add the standard Apache attribution comment where needed. Could you check all of them? ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/ControlRequestSendThread.java: ########## @@ -0,0 +1,204 @@ +/* + * 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.coordinator.channel; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metrics.Counter; +import org.apache.fluss.metrics.DescriptiveStatisticsHistogram; +import org.apache.fluss.metrics.Histogram; +import org.apache.fluss.metrics.MetricNames; +import org.apache.fluss.metrics.groups.MetricGroup; +import org.apache.fluss.rpc.gateway.TabletServerGateway; +import org.apache.fluss.rpc.messages.ApiMessage; +import org.apache.fluss.rpc.messages.StopReplicaRequest; +import org.apache.fluss.utils.concurrent.ShutdownableThread; + +import java.util.Optional; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.IntSupplier; +import java.util.function.Supplier; + +/** + * Per-tablet-server sender thread that drains the control-plane request queue and retries on + * transient RPC failures. Mirrors Kafka's {@code RequestSendThread} + * (ControllerChannelManager.scala:226-300). Review Comment: There's no need to emphasize this here. If you want to indicate that you referenced an Apache open source project, please use a more standard comment format — refer to `LogManager` for an example: ``` /* This file is based on source code of Apache Kafka Project (https://kafka.apache.org/), licensed by the Apache * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for * additional information regarding copyright ownership. */ ``` ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/TabletServerChannelState.java: ########## @@ -0,0 +1,93 @@ +/* + * 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.coordinator.channel; + +import org.apache.fluss.cluster.ServerNode; +import org.apache.fluss.metrics.groups.MetricGroup; + +import javax.annotation.concurrent.ThreadSafe; + +import java.util.concurrent.BlockingQueue; + +/** + * Per-tablet-server channel state held by {@link + * org.apache.fluss.server.coordinator.CoordinatorChannelManager}: the queue, the sender thread, and + * the metric handles needed to deregister the per-TS metrics on remove. + * + * <p>Mirrors Kafka's {@code ControllerBrokerStateInfo} (ControllerChannelManager.scala:768-774). + * The {@code networkClient} field is intentionally absent — {@code RpcGatewayManager} owns the + * gateway exactly the way Kafka's controller owns the {@code NetworkClient}, so we do not duplicate + * it on the state. Three Fluss-specific additions surface behavior that has no Kafka analog: + * + * <ul> + * <li>{@code retryCount} — counts §5.4 retry attempts. Review Comment: What's this tag means: `§5.4` ? Pls check all the code refer to these unknown tag. ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java: ########## @@ -919,43 +959,93 @@ private void processDropPartition(DropPartitionEvent dropPartitionEvent) { dropTableInfo.getTablePath(), tableId, tablePartition.getPartitionId()); } + /** + * Handles StopReplica deletion responses from the sender thread. Partitions per-bucket results + * into succeeded/failed sets, then moves replicas to the appropriate state. + * + * <p>Mirrors Kafka's {@code processTopicDeletionStopReplicaResponseReceived} + * (KafkaController.scala:1451-1471). Fluss's StopReplicaResponse has no top-level error code + * (FlussApi.proto:373-375), so the Kafka "requestError != NONE → all in error" branch does not + * apply. Each bucket's error code is evaluated individually. + * + * <p>No {@code isActive} guard is needed (unlike Kafka :1454): {@code + * coordinatorEventManager.close()} in {@link #shutdown()} joins the event thread before {@code + * coordinatorChannelManager.shutdown()}, so no event runs after resignation. + */ private void processDeleteReplicaResponseReceived( DeleteReplicaResponseReceivedEvent deleteReplicaResponseReceivedEvent) { List<DeleteReplicaResultForBucket> deleteReplicaResultForBuckets = deleteReplicaResponseReceivedEvent.getDeleteReplicaResults(); - Set<TableBucketReplica> failDeletedReplicas = new HashSet<>(); - Set<TableBucketReplica> successDeletedReplicas = new HashSet<>(); - for (DeleteReplicaResultForBucket deleteReplicaResultForBucket : - deleteReplicaResultForBuckets) { - TableBucketReplica tableBucketReplica = - deleteReplicaResultForBucket.getTableBucketReplica(); - if (deleteReplicaResultForBucket.succeeded()) { - successDeletedReplicas.add(tableBucketReplica); + Set<TableBucketReplica> succeeded = new HashSet<>(); + Set<TableBucketReplica> failed = new HashSet<>(); + for (DeleteReplicaResultForBucket result : deleteReplicaResultForBuckets) { + if (result.succeeded()) { + succeeded.add(result.getTableBucketReplica()); } else { - failDeletedReplicas.add(tableBucketReplica); + failed.add(result.getTableBucketReplica()); } } - // clear the fail deleted number for the success deleted replicas - coordinatorContext.clearFailDeleteNumbers(successDeletedReplicas); - // pick up the replicas to retry delete and replicas that considered as success delete - Tuple2<Set<TableBucketReplica>, Set<TableBucketReplica>> - retryDeleteAndSuccessDeleteReplicas = - coordinatorContext.retryDeleteAndSuccessDeleteReplicas(failDeletedReplicas); + if (!failed.isEmpty()) { + failReplicaDeletion(failed); + } + if (!succeeded.isEmpty()) { + completeReplicaDeletion(succeeded); + } + } - // transmit to deletion started for retry delete replicas - replicaStateMachine.handleStateChanges( - retryDeleteAndSuccessDeleteReplicas.f0, ReplicaDeletionStarted); + /** + * Mirrors Kafka's {@code TopicDeletionManager.failReplicaDeletion} Review Comment: ditto ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/TabletServerChannelState.java: ########## @@ -0,0 +1,93 @@ +/* + * 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.coordinator.channel; + +import org.apache.fluss.cluster.ServerNode; +import org.apache.fluss.metrics.groups.MetricGroup; + +import javax.annotation.concurrent.ThreadSafe; + +import java.util.concurrent.BlockingQueue; + +/** + * Per-tablet-server channel state held by {@link + * org.apache.fluss.server.coordinator.CoordinatorChannelManager}: the queue, the sender thread, and + * the metric handles needed to deregister the per-TS metrics on remove. + * + * <p>Mirrors Kafka's {@code ControllerBrokerStateInfo} (ControllerChannelManager.scala:768-774). Review Comment: ditto ########## fluss-server/src/test/java/org/apache/fluss/server/coordinator/channel/ControlRequestSendThreadTest.java: ########## @@ -0,0 +1,349 @@ +/* + * 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.coordinator.channel; + +import org.apache.fluss.config.ConfigOptions; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metrics.Counter; +import org.apache.fluss.metrics.Gauge; +import org.apache.fluss.metrics.MetricNames; +import org.apache.fluss.metrics.groups.MetricGroup; +import org.apache.fluss.metrics.util.TestMetricGroup; +import org.apache.fluss.rpc.gateway.TabletServerGateway; +import org.apache.fluss.rpc.messages.StopReplicaRequest; +import org.apache.fluss.rpc.messages.StopReplicaResponse; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Proxy; +import java.util.Optional; +import java.util.concurrent.BlockingQueue; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.LinkedBlockingQueue; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Function; +import java.util.function.IntSupplier; +import java.util.function.Supplier; + +import static org.assertj.core.api.Assertions.assertThat; + +/** Tests for {@link ControlRequestSendThread}. */ +class ControlRequestSendThreadTest { + + private static final int TABLET_SERVER_ID = 1; + private static final int EPOCH = 1; + + private ControlRequestSendThread thread; + + @AfterEach + void afterEach() throws InterruptedException { + if (thread != null) { + thread.initiateShutdown(); + thread.awaitShutdown(); + } + } + + /** U1: happy path — single item dequeued, dispatched, callback invoked. */ + @Test + void testHappyPath() throws Exception { + BlockingQueue<QueueItem> queue = new LinkedBlockingQueue<>(); + MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup(); + + StopReplicaResponse response = new StopReplicaResponse(); + AtomicInteger invocationCount = new AtomicInteger(0); + TabletServerGateway gateway = + stubGateway( + req -> { + invocationCount.incrementAndGet(); + return CompletableFuture.completedFuture(response); + }); + + AtomicReference<Object> callbackResponse = new AtomicReference<>(); + CountDownLatch callbackLatch = new CountDownLatch(1); + + thread = createThread(queue, () -> Optional.of(gateway), () -> EPOCH, metricGroup); + thread.start(); + + queue.put( + new QueueItem( + ApiKey.STOP_REPLICA, + new StopReplicaRequest(), + (resp, err) -> { + callbackResponse.set(resp); + callbackLatch.countDown(); + }, + EPOCH, + System.currentTimeMillis(), + null)); + + assertThat(callbackLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(callbackResponse.get()).isSameAs(response); + assertThat(getCounter(metricGroup, MetricNames.SENDER_RETRY_COUNT).getCount()).isEqualTo(0); + assertThat(getCounter(metricGroup, MetricNames.SENDER_STALE_DROP_COUNT).getCount()) + .isEqualTo(0); + assertThat(getGaugeValue(metricGroup, MetricNames.SENDER_ALIVE)).isEqualTo(1); + assertThat(invocationCount.get()).isEqualTo(1); + } + + /** U2: retry then succeed — gateway fails once, succeeds on second attempt. */ + @Test + void testRetryThenSucceed() throws Exception { + BlockingQueue<QueueItem> queue = new LinkedBlockingQueue<>(); + MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup(); + + StopReplicaResponse response = new StopReplicaResponse(); + AtomicInteger callCount = new AtomicInteger(0); + TabletServerGateway gateway = + stubGateway( + req -> { + if (callCount.incrementAndGet() == 1) { + CompletableFuture<StopReplicaResponse> fail = + new CompletableFuture<>(); + fail.completeExceptionally(new RuntimeException("transient error")); + return fail; + } + return CompletableFuture.completedFuture(response); + }); + + CountDownLatch callbackLatch = new CountDownLatch(1); + + thread = createThread(queue, () -> Optional.of(gateway), () -> EPOCH, metricGroup); + thread.start(); + + queue.put( + new QueueItem( + ApiKey.STOP_REPLICA, + new StopReplicaRequest(), + (resp, err) -> callbackLatch.countDown(), + EPOCH, + System.currentTimeMillis(), + null)); + + assertThat(callbackLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(getCounter(metricGroup, MetricNames.SENDER_RETRY_COUNT).getCount()) + .isGreaterThanOrEqualTo(1); + } + + /** U4: stale-epoch drop — item with older epoch is dropped, staleDropCount increments. */ + @Test + void testStaleEpochDrop() throws Exception { + BlockingQueue<QueueItem> queue = new LinkedBlockingQueue<>(); + MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup(); + + StopReplicaResponse response = new StopReplicaResponse(); + TabletServerGateway gateway = + stubGateway(req -> CompletableFuture.completedFuture(response)); + + CountDownLatch sentinelLatch = new CountDownLatch(1); + + thread = createThread(queue, () -> Optional.of(gateway), () -> 5, metricGroup); + thread.start(); + + // stale item (epoch 1 < current 5) + queue.put( + new QueueItem( + ApiKey.STOP_REPLICA, + new StopReplicaRequest(), + (resp, err) -> {}, + 1, + System.currentTimeMillis(), + null)); + + // sentinel item with current epoch to detect when stale item was processed + queue.put( + new QueueItem( + ApiKey.STOP_REPLICA, + new StopReplicaRequest(), + (resp, err) -> sentinelLatch.countDown(), + 5, + System.currentTimeMillis(), + null)); + + assertThat(sentinelLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(getCounter(metricGroup, MetricNames.SENDER_STALE_DROP_COUNT).getCount()) + .isEqualTo(1); + } + + /** U5: gateway absent then present — retries until gateway becomes available. */ + @Test + void testGatewayAbsentThenPresent() throws Exception { + BlockingQueue<QueueItem> queue = new LinkedBlockingQueue<>(); + MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup(); + + StopReplicaResponse response = new StopReplicaResponse(); + TabletServerGateway gateway = + stubGateway(req -> CompletableFuture.completedFuture(response)); + + AtomicInteger gatewayCallCount = new AtomicInteger(0); + Supplier<Optional<TabletServerGateway>> gatewaySupplier = + () -> { + if (gatewayCallCount.incrementAndGet() <= 3) { + return Optional.empty(); + } + return Optional.of(gateway); + }; + + CountDownLatch callbackLatch = new CountDownLatch(1); + + thread = createThread(queue, gatewaySupplier, () -> EPOCH, metricGroup); + thread.start(); + + queue.put( + new QueueItem( + ApiKey.STOP_REPLICA, + new StopReplicaRequest(), + (resp, err) -> callbackLatch.countDown(), + EPOCH, + System.currentTimeMillis(), + null)); + + assertThat(callbackLatch.await(5, TimeUnit.SECONDS)).isTrue(); + assertThat(getCounter(metricGroup, MetricNames.SENDER_RETRY_COUNT).getCount()) + .isGreaterThanOrEqualTo(3); + } + + /** U7: shutdown cancels in-flight — shutdown completes quickly, callback never invoked. */ + @Test + void testShutdownCancelsInFlight() throws Exception { + BlockingQueue<QueueItem> queue = new LinkedBlockingQueue<>(); + MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup(); + + CountDownLatch invokedLatch = new CountDownLatch(1); + CompletableFuture<StopReplicaResponse> neverCompleteFuture = new CompletableFuture<>(); + TabletServerGateway gateway = + stubGateway( + req -> { + invokedLatch.countDown(); + return neverCompleteFuture; + }); + + AtomicInteger callbackCount = new AtomicInteger(0); + + thread = createThread(queue, () -> Optional.of(gateway), () -> EPOCH, metricGroup); + thread.start(); + + queue.put( + new QueueItem( + ApiKey.STOP_REPLICA, + new StopReplicaRequest(), + (resp, err) -> callbackCount.incrementAndGet(), + EPOCH, + System.currentTimeMillis(), + null)); + + assertThat(invokedLatch.await(5, TimeUnit.SECONDS)).isTrue(); + + long shutdownStart = System.currentTimeMillis(); + thread.shutdown(); + long shutdownDuration = System.currentTimeMillis() - shutdownStart; + + assertThat(shutdownDuration).isLessThan(2000); + assertThat(callbackCount.get()).isEqualTo(0); + assertThat(getGaugeValue(metricGroup, MetricNames.SENDER_ALIVE)).isEqualTo(0); + thread = null; + } + + /** U8: callback throws after successful send — thread survives, no re-send. */ + @Test + void testCallbackThrowsDoesNotRetry() throws Exception { + BlockingQueue<QueueItem> queue = new LinkedBlockingQueue<>(); + MetricGroup metricGroup = TestMetricGroup.createTestMetricGroup(); + + StopReplicaResponse response = new StopReplicaResponse(); + AtomicInteger invocationCount = new AtomicInteger(0); + TabletServerGateway gateway = + stubGateway( + req -> { + invocationCount.incrementAndGet(); + return CompletableFuture.completedFuture(response); + }); + + CountDownLatch secondCallbackLatch = new CountDownLatch(1); + + thread = createThread(queue, () -> Optional.of(gateway), () -> EPOCH, metricGroup); + thread.start(); + + // first item: callback throws + queue.put( + new QueueItem( + ApiKey.STOP_REPLICA, + new StopReplicaRequest(), + (resp, err) -> { + throw new RuntimeException("buggy callback"); + }, + EPOCH, + System.currentTimeMillis(), + null)); + + // second item: proves thread survived + queue.put( + new QueueItem( + ApiKey.STOP_REPLICA, + new StopReplicaRequest(), + (resp, err) -> secondCallbackLatch.countDown(), + EPOCH, + System.currentTimeMillis(), + null)); + + assertThat(secondCallbackLatch.await(5, TimeUnit.SECONDS)).isTrue(); + // gateway invoked exactly twice (once per item, no re-send from callback failure) + assertThat(invocationCount.get()).isEqualTo(2); + assertThat(getCounter(metricGroup, MetricNames.SENDER_RETRY_COUNT).getCount()).isEqualTo(0); + assertThat(getGaugeValue(metricGroup, MetricNames.SENDER_ALIVE)).isEqualTo(1); + } + + @SuppressWarnings("unchecked") Review Comment: Redundant suppression, can be remove. ########## fluss-server/src/main/java/org/apache/fluss/server/coordinator/channel/TabletServerChannelState.java: ########## @@ -0,0 +1,93 @@ +/* + * 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.coordinator.channel; + +import org.apache.fluss.cluster.ServerNode; +import org.apache.fluss.metrics.groups.MetricGroup; + +import javax.annotation.concurrent.ThreadSafe; + +import java.util.concurrent.BlockingQueue; + +/** + * Per-tablet-server channel state held by {@link + * org.apache.fluss.server.coordinator.CoordinatorChannelManager}: the queue, the sender thread, and + * the metric handles needed to deregister the per-TS metrics on remove. + * + * <p>Mirrors Kafka's {@code ControllerBrokerStateInfo} (ControllerChannelManager.scala:768-774). + * The {@code networkClient} field is intentionally absent — {@code RpcGatewayManager} owns the + * gateway exactly the way Kafka's controller owns the {@code NetworkClient}, so we do not duplicate + * it on the state. Three Fluss-specific additions surface behavior that has no Kafka analog: + * + * <ul> + * <li>{@code retryCount} — counts §5.4 retry attempts. + * <li>{@code staleDropCount} — counts §5.4.1 stale-epoch drops at dequeue. + * <li>{@code aliveGauge} — presence indicator for the per-TS sender (§8.5 Scenario E). + * </ul> + * + * <p>Each field is final and assigned once; mutable concurrent state is encapsulated in the queue, + * the thread, and the metric handles themselves. + */ +@ThreadSafe +public final class TabletServerChannelState { + + private final int tabletServerId; + private final ServerNode serverNode; + private final BlockingQueue<QueueItem> queue; + private final ControlRequestSendThread sendThread; + + /** + * The per-TS child {@link MetricGroup}. Held so {@code removeTabletServer} can {@code close()} + * it and deregister all five per-TS metrics in one shot — mirrors Kafka's two {@code + * metricsGroup.removeMetric(...)} calls at ControllerChannelManager.scala:203-204. + */ + private final MetricGroup metricGroup; + + public TabletServerChannelState( + int tabletServerId, + ServerNode serverNode, + BlockingQueue<QueueItem> queue, + ControlRequestSendThread sendThread, + MetricGroup metricGroup) { + this.tabletServerId = tabletServerId; + this.serverNode = serverNode; + this.queue = queue; + this.sendThread = sendThread; + this.metricGroup = metricGroup; + } + + public int getTabletServerId() { + return tabletServerId; + } + + public ServerNode getServerNode() { Review Comment: No user. remove it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
