RongtongJin commented on code in PR #11046:
URL: https://github.com/apache/rocketmq/pull/11046#discussion_r4227688656
##########
controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/DefaultBrokerHeartbeatManager.java:
##########
@@ -111,53 +112,70 @@ public void onBrokerHeartbeat(String clusterName, String
brokerName, String brok
Long timeoutMillis, Channel channel, Integer epoch, Long maxOffset,
Long confirmOffset,
Integer electionPriority) {
BrokerIdentityInfo brokerIdentityInfo = new
BrokerIdentityInfo(clusterName, brokerName, brokerId);
- BrokerLiveInfo prev = this.brokerLiveTable.get(brokerIdentityInfo);
int realEpoch = Optional.ofNullable(epoch).orElse(-1);
long realBrokerId = Optional.ofNullable(brokerId).orElse(-1L);
long realMaxOffset = Optional.ofNullable(maxOffset).orElse(-1L);
long realConfirmOffset =
Optional.ofNullable(confirmOffset).orElse(-1L);
long realTimeoutMillis =
Optional.ofNullable(timeoutMillis).orElse(DEFAULT_BROKER_CHANNEL_EXPIRED_TIME);
int realElectionPriority =
Optional.ofNullable(electionPriority).orElse(Integer.MAX_VALUE);
- if (null == prev) {
- this.brokerLiveTable.put(brokerIdentityInfo,
- new BrokerLiveInfo(brokerName,
+ // compute() keeps the create-or-update decision and the channel
rebinding atomic per
+ // broker identity, so onBrokerChannelClose can never observe a
half-updated entry: a
+ // heartbeat that rebinds the entry and a close of the previous
channel serialize on the
+ // same key instead of racing between the match and the removal.
+ this.brokerLiveTable.compute(brokerIdentityInfo, (key, prev) -> {
+ if (prev == null) {
+ log.info("new broker registered, {}, brokerId:{}",
brokerIdentityInfo, realBrokerId);
+ return new BrokerLiveInfo(brokerName,
brokerAddr,
realBrokerId,
System.currentTimeMillis(),
realTimeoutMillis,
channel,
realEpoch,
realMaxOffset,
- realElectionPriority));
- log.info("new broker registered, {}, brokerId:{}",
brokerIdentityInfo, realBrokerId);
- } else {
+ realElectionPriority);
+ }
prev.setLastUpdateTimestamp(System.currentTimeMillis());
Review Comment:
[P1] Make expiry revalidation/removal atomic with heartbeat rebinding
The per-key compute protects this update against onBrokerChannelClose, but
scanNotActiveBroker still reads the timestamp, checks expiry, reads the
channel, and removes the entry outside that atomic operation. A concrete
interleaving is:
1. The scanner reads the old lastUpdateTimestamp and pauses.
2. A heartbeat on newChannel refreshes the timestamp and rebinds this same
BrokerLiveInfo.
3. The scanner resumes, decides it is expired using the old timestamp, reads
newChannel from the updated object, removes the live entry, and closes the
healthy new connection.
I reproduced this with JDK 8 by pausing getLastUpdateTimestamp after
capturing its old value, completing the replacement heartbeat, then resuming
the scan. The assertion that newChannel remains active passes with the base
manager and fails at this head. The expiry/removal race predates this PR, but
closing the newly rebound connection is an additional consequence of this
change.
Please have the scan recheck the current timestamp/timeout and remove the
entry within computeIfPresent for the same identity, capture the channel of the
entry actually removed, and close that captured channel outside the operation.
Add a coordinated regression test for this order; remove(key, value) alone is
insufficient while heartbeats mutate the same value object.
##########
controller/src/main/java/org/apache/rocketmq/controller/impl/heartbeat/DefaultBrokerHeartbeatManager.java:
##########
@@ -111,53 +112,70 @@ public void onBrokerHeartbeat(String clusterName, String
brokerName, String brok
Long timeoutMillis, Channel channel, Integer epoch, Long maxOffset,
Long confirmOffset,
Integer electionPriority) {
BrokerIdentityInfo brokerIdentityInfo = new
BrokerIdentityInfo(clusterName, brokerName, brokerId);
- BrokerLiveInfo prev = this.brokerLiveTable.get(brokerIdentityInfo);
int realEpoch = Optional.ofNullable(epoch).orElse(-1);
long realBrokerId = Optional.ofNullable(brokerId).orElse(-1L);
long realMaxOffset = Optional.ofNullable(maxOffset).orElse(-1L);
long realConfirmOffset =
Optional.ofNullable(confirmOffset).orElse(-1L);
long realTimeoutMillis =
Optional.ofNullable(timeoutMillis).orElse(DEFAULT_BROKER_CHANNEL_EXPIRED_TIME);
int realElectionPriority =
Optional.ofNullable(electionPriority).orElse(Integer.MAX_VALUE);
- if (null == prev) {
- this.brokerLiveTable.put(brokerIdentityInfo,
- new BrokerLiveInfo(brokerName,
+ // compute() keeps the create-or-update decision and the channel
rebinding atomic per
+ // broker identity, so onBrokerChannelClose can never observe a
half-updated entry: a
+ // heartbeat that rebinds the entry and a close of the previous
channel serialize on the
+ // same key instead of racing between the match and the removal.
+ this.brokerLiveTable.compute(brokerIdentityInfo, (key, prev) -> {
+ if (prev == null) {
+ log.info("new broker registered, {}, brokerId:{}",
brokerIdentityInfo, realBrokerId);
+ return new BrokerLiveInfo(brokerName,
brokerAddr,
realBrokerId,
System.currentTimeMillis(),
realTimeoutMillis,
channel,
realEpoch,
realMaxOffset,
- realElectionPriority));
- log.info("new broker registered, {}, brokerId:{}",
brokerIdentityInfo, realBrokerId);
- } else {
+ realElectionPriority);
+ }
prev.setLastUpdateTimestamp(System.currentTimeMillis());
prev.setHeartbeatTimeoutMillis(realTimeoutMillis);
prev.setElectionPriority(realElectionPriority);
+ // Rebind the live entry to the current channel, otherwise a close
event of a stale
+ // channel would evict a broker that is already alive on a new
channel
+ if (channel != null && channel != prev.getChannel()) {
+ prev.setChannel(channel);
Review Comment:
[P1] Do not let a late heartbeat restore an already closed channel
This condition also accepts a heartbeat whose channel has already closed. A
heartbeat request from oldChannel can be dispatched before disconnection but
finish after a heartbeat on newChannel: controller requests use the shared
request executor, RequestTask.run does not check channel liveness, and close
events run on a separate NettyEventExecutor.
The resulting order is: an old heartbeat is queued; oldChannel closes;
newChannel's heartbeat updates the live entry; the queued old heartbeat runs
and this assignment changes the channel back to oldChannel; the delayed old
close callback then removes the broker even though newChannel is still active.
I reproduced that order with an actual RequestTask and EmbeddedChannels on this
head: the new connection remains active, but the live-table entry disappears.
This is a remaining instance of the original failure, rather than a separate
newly introduced bug. compute serializes the map operations but cannot
establish which connection is current. Please reject heartbeats from closed
channels inside the per-identity atomic update, covering creation and timestamp
refresh as well as rebinding, and add a deterministic test where the old
request executes after the new heartbeat. If two active connections may
overlap, their ordering also needs an explicit policy; the broker epoch alone
is not a connection generation.
--
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]
