qiuyanjun888 commented on code in PR #18416:
URL:
https://github.com/apache/dolphinscheduler/pull/18416#discussion_r4239991952
##########
dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/src/main/java/org/apache/dolphinscheduler/plugin/registry/jdbc/server/JdbcRegistryServer.java:
##########
@@ -341,38 +366,69 @@ private void refreshClientsHeartbeat() {
}
JdbcRegistryClientHeartbeatDTO clone =
jdbcRegistryClientHeartbeatDTO.clone();
clone.setLastHeartbeatTime(now);
-
jdbcRegistryClientRepository.updateById(jdbcRegistryClientHeartbeatDTO);
+ if (!jdbcRegistryClientRepository.updateById(clone)) {
+ log.error("The client heartbeat has expired: {}",
jdbcRegistryClientHeartbeatDTO.getId());
+ throw new IllegalStateException(
+ "The client heartbeat record no longer exists: " +
jdbcRegistryClientHeartbeatDTO.getId());
+ }
jdbcRegistryClientHeartbeatDTO.setLastHeartbeatTime(clone.getLastHeartbeatTime());
}
- if (jdbcRegistryServerState == JdbcRegistryServerState.SUSPENDED) {
- jdbcRegistryServerState = JdbcRegistryServerState.STARTED;
- doTriggerReconnectedListener();
+ currentState = serverState.get();
+ boolean reconnected = currentState ==
JdbcRegistryServerState.SUSPENDED;
+ if (reconnected) {
+ if
(!serverState.compareAndSet(JdbcRegistryServerState.SUSPENDED,
JdbcRegistryServerState.STARTED)) {
+ log.debug("Failed to reconnect JdbcRegistryServer; current
state is {}", serverState.get());
+ return;
+ }
+ } else if (currentState != JdbcRegistryServerState.STARTED) {
+ return;
+ }
+ // Serialize heartbeat side effects with close(), even if close
wins after the state transition.
+ synchronized (this) {
+ if (serverState.get() != JdbcRegistryServerState.STARTED) {
+ return;
+ }
+ lastSuccessHeartbeat = now;
+ if (reconnected) {
+ doTriggerReconnectedListener();
+ }
Review Comment:
Yes. The extra synchronization was added only to prevent an in-flight
heartbeat from racing with service shutdown. Since close() is only invoked
during application shutdown in production, I will remove that shutdown-specific
coordination and keep CAS only for the heartbeat state transitions. The
conditional heartbeat deletion will remain because it protects the real
purge-versus-renewal race.
--
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]
