loserwang1024 commented on code in PR #3400:
URL: https://github.com/apache/fluss/pull/3400#discussion_r3338499807
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorContext.java:
##########
@@ -106,6 +106,13 @@ public class CoordinatorContext {
*/
private final Map<Integer, Set<TableBucket>> replicasOnOffline = new
HashMap<>();
+ /**
+ * Tracks leader buckets that are currently inactive — a leader is
inactive from the moment we
+ * send NotifyLeaderAndIsr until the target server responds successfully
confirming it is the
+ * leader. Also inactive when leader == NO_LEADER.
+ */
+ private final Set<TableBucket> inactiveLeaderBuckets = new HashSet<>();
Review Comment:
Naming: inactiveLeaderBuckets → pendingLeaderActivationBuckets
The name over-claims. The set is only populated in
sendNotifyLeaderAndIsrRequest — buckets where we've sent a NotifyLeaderAndIsr
and await leadership confirmation. Buckets leaderless for other reasons (server
offline, rebalance with no new leader yet) aren't in it, though their leader is
effectively inactive. So it really means "leader change dispatched, awaiting
confirmation" — pendingLeaderActivationBuckets fits. Avoid pendingIsr...: this
has nothing to do with ISR (that's a separate dimension driving YELLOW).
Bigger issue: "active leader" is defined twice. computeClusterHealth checks
leader != NO_LEADER && liveServers.contains(leader) && !inSet, but the helper
only checks the set:
```java
public boolean isLeaderActive(TableBucket bucket) {
return !inactiveLeaderBuckets.contains(bucket); // ignores liveServers /
NO_LEADER
}
```
So for a leader on a dead server, isLeaderActive() says true while
computeClusterHealth() says RED — inconsistent, and a trap for future reuse.
Suggestion: rename the set + make isLeaderActive the single source of truth:
```java
public boolean isLeaderActive(TableBucket tb) {
return getBucketLeaderAndIsr(tb)
.map(lai -> lai.leader() != LeaderAndIsr.NO_LEADER
&& liveTabletServers.containsKey(lai.leader())
&& !pendingLeaderActivationBuckets.contains(tb))
.orElse(false);
}
```
Then computeClusterHealth just calls ctx.isLeaderActive(tb).
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorService.java:
##########
@@ -291,6 +294,62 @@ public CoordinatorService(
this.coordinatorLeaderElection = coordinatorLeaderElection;
}
+ @Override
+ public CompletableFuture<GetClusterHealthResponse> getClusterHealth(
+ GetClusterHealthRequest request) {
+ AccessContextEvent<GetClusterHealthResponse> event =
+ new
AccessContextEvent<>(CoordinatorService::computeClusterHealth);
+ eventManagerSupplier.get().put(event);
+ return event.getResultFuture();
+ }
+
+ @VisibleForTesting
+ static GetClusterHealthResponse computeClusterHealth(CoordinatorContext
ctx) {
+ GetClusterHealthResponse response = new GetClusterHealthResponse();
+
+ int numReplicas = 0;
+ int inSyncReplicas = 0;
+ int numLeaderReplicas = 0;
+ int activeLeaderReplicas = 0;
+ Set<TableBucket> inactiveLeaders = ctx.getInactiveLeaderBuckets();
+
+ for (TableBucket tb : ctx.getAllBuckets()) {
+ List<Integer> assignment = ctx.getAssignment(tb);
+ numReplicas += assignment.size();
+ numLeaderReplicas++;
+
+ Optional<LeaderAndIsr> laiOpt = ctx.getBucketLeaderAndIsr(tb);
+ if (laiOpt.isPresent()) {
+ LeaderAndIsr lai = laiOpt.get();
+ inSyncReplicas += lai.isr().size();
+ if (lai.leader() != LeaderAndIsr.NO_LEADER
+ && ctx.getLiveTabletServers().containsKey(lai.leader())
+ && !inactiveLeaders.contains(tb)) {
+ activeLeaderReplicas++;
+ }
+ }
+ }
+
+ // PbClusterHealthStatus: GREEN=0, YELLOW=1, RED=2, UNKNOWN=3
+ int status;
+ if (numLeaderReplicas == 0) {
Review Comment:
This can merge with else {
status = 0; // GREEN
}
##########
fluss-server/src/main/java/org/apache/fluss/server/RpcServiceBase.java:
##########
@@ -574,6 +576,18 @@ public CompletableFuture<DescribeClusterConfigsResponse>
describeClusterConfigs(
new
DescribeClusterConfigsResponse().addAllConfigs(toPbConfigEntries(configs)));
}
+ @Override
+ public CompletableFuture<GetClusterHealthResponse> getClusterHealth(
Review Comment:
Why not move to CoordinatorGateway? Could tablet server support it?
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorEventProcessor.java:
##########
@@ -974,10 +975,17 @@ private void
processNotifyLeaderAndIsrResponseReceivedEvent(
offlineReplicas.add(
new TableBucketReplica(
notifyLeaderAndIsrResultForBucket.getTableBucket(), serverId));
+ } else {
+
succeededBuckets.add(notifyLeaderAndIsrResultForBucket.getTableBucket());
+ }
+ }
+ for (TableBucket tb : succeededBuckets) {
+ Optional<LeaderAndIsr> laiOpt =
coordinatorContext.getBucketLeaderAndIsr(tb);
+ if (laiOpt.isPresent() && laiOpt.get().leader() == serverId) {
+ coordinatorContext.markLeaderActive(tb);
}
}
if (!offlineReplicas.isEmpty()) {
- // trigger replicas to offline
Review Comment:
why remove this comment?
##########
helm/templates/sts-coordinator.yaml:
##########
@@ -122,7 +124,7 @@ spec:
echo "bind.listeners: ${BIND_LISTENERS}" >>
$FLUSS_HOME/conf/server.yaml && \
echo "advertised.listeners: ${ADVERTISED_LISTENERS}" >>
$FLUSS_HOME/conf/server.yaml && \
- bin/coordinator-server.sh start-foreground
+ exec bin/coordinator-server.sh start-foreground
Review Comment:
why need to change it?
##########
fluss-server/src/main/java/org/apache/fluss/server/coordinator/CoordinatorRequestBatch.java:
##########
@@ -422,12 +433,20 @@ private void sendNotifyLeaderAndIsrRequest(int
coordinatorEpoch) {
"Failed to send notify leader and isr
request to tablet server {}.",
serverId,
throwable);
- // todo: in FLUSS-55886145, we will introduce a
sender thread to send
- // the request, and retry if encounter any error;
It may happens that
- // the tablet server is offline and will always
got error. But,
- // coordinator will remove the sender for the
tablet server and mark all
- // replica in the tablet server as offline. so, in
here, if encounter
- // any error, we just ignore it.
+ // Treat all buckets as failed — clears pending
state and triggers
+ // re-election via onReplicaBecomeOffline.
+ List<NotifyLeaderAndIsrResultForBucket>
failedResults =
+ new ArrayList<>();
+ ApiError sendError =
Review Comment:
Currently, after this modification, even a network timeout exception, the
CoordinatorEventProcessor#processNotifyLeaderAndIsrResponseReceivedEvent will
mark the server as offlineReplicas
won't is any problem?
<img width="1062" height="280" alt="Image"
src="https://github.com/user-attachments/assets/ecbd5a3a-4568-40ed-8c5d-d8871bd94fd5";
/>
--
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]
