fxbing opened a new issue, #3698: URL: https://github.com/apache/fluss/issues/3698
### Search before asking - [x] I searched in the [issues](https://github.com/apache/fluss/issues) and found nothing similar. ### Fluss version main (development) ### Please describe the bug 🐞 A newly elected leader may remove healthy followers from ISR as soon as the first record is appended, before those followers send their next fetch. A repeated `NotifyLeaderAndIsr` request with the same leader epoch may cause the same problem. This can appear as an `isrShrinksPerSecond` spike when traffic starts on an otherwise healthy table, followed by an ISR expansion after the next follower fetch. #### How to reproduce 1. Make TabletServer 1 the leader with replicas and ISR `[1, 2, 3]`. 2. Append one record before followers 2 and 3 send their next fetch. 3. Trigger the ISR expiration check. Expected: the ISR remains `[1, 2, 3]` during the normal `log.replica.max-lag-time` grace period. Actual: the ISR shrinks to `[1]`. Followers 2 and 3 can rejoin after their next fetch. #### Root cause [`Replica.makeLeader()`](https://github.com/apache/fluss/blob/d7057f52f5ef9a9e61d0b0718f23734eff9f2302/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java#L419-L469) calls `updateAssignmentAndIsr()` before checking the leader epoch. [`updateAssignmentAndIsr()`](https://github.com/apache/fluss/blob/d7057f52f5ef9a9e61d0b0718f23734eff9f2302/fluss-server/src/main/java/org/apache/fluss/server/replica/Replica.java#L1219-L1247) unconditionally replaces every `FollowerReplica`: ```java followerReplicasMap.put(replica, new FollowerReplica(replica, tableBucket)); ``` This discards the follower LEO and caught-up timestamps. A newly created `FollowerReplica` has an unknown LEO and `lastCaughtUpTimeMs = 0`, so it becomes immediately eligible for removal from ISR once the leader LEO advances. [`FollowerReplica.resetFollowerReplicaState()`](https://github.com/apache/fluss/blob/d7057f52f5ef9a9e61d0b0718f23734eff9f2302/fluss-server/src/main/java/org/apache/fluss/server/replica/FollowerReplica.java#L98-L128) already defines the intended initialization for leader election and re-election, including setting `lastCaughtUpTimeMs` to the current time for followers in ISR, but it currently has no production caller. ### Solution Follow the replica lifecycle used by Kafka: - Keep existing `FollowerReplica` objects for unchanged assignments, for example with `computeIfAbsent`, so repeated notifications do not discard their fetch state. - When the leader epoch advances, reset each follower through the existing `resetFollowerReplicaState(...)` method. Followers in the committed ISR should receive the normal lag grace period. - Add regression tests for an immediate first write after leader election and for a repeated notification with the same leader epoch. Kafka similarly [resets remote replica state when the leader epoch changes](https://github.com/apache/kafka/blob/56455183deefcd50f6b76f5dea5e84dd884274cb/core/src/main/scala/kafka/cluster/Partition.scala#L633-L660) and [retains existing replica objects with `computeIfAbsent`](https://github.com/apache/kafka/blob/56455183deefcd50f6b76f5dea5e84dd884274cb/core/src/main/scala/kafka/cluster/Partition.scala#L843-L849). ### Are you willing to submit a PR? - [x] I'm willing to submit a PR! -- 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]
