RockteMQ-AI commented on code in PR #433:
URL: https://github.com/apache/rocketmq-connect/pull/433#discussion_r3839522633
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerConnector.java:
##########
@@ -369,10 +369,18 @@ public String toString() {
return sb;
}
- private enum State {
+ public enum State {
Review Comment:
The `state` field is not `volatile`, yet `getState()` is now called from the
`StateMachineService` thread (in `redressRunningConnectors()`) while `state` is
written from the connector's own thread (e.g., in `doStart()`, `onFailure()`,
`pause()`). This is a data race under the Java Memory Model — reads may return
stale values. The field should be declared `volatile` to guarantee cross-thread
visibility.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java:
##########
@@ -564,7 +565,8 @@ public Set<Runnable> getCleanedStoppedTasks() {
}
public void maintainConnectorState() {
-
+ // STEP 1: redress running connectors status
Review Comment:
No test coverage is provided for `redressRunningConnectors()`. The existing
`redressRunningStatus()` for tasks also lacks tests. Given that this method
modifies distributed connector state and interacts with
`stateManagementService`, at minimum a unit test should verify: (1) UNASSIGNED
+ STARTED connector gets redressed to RUNNING, (2) already-RUNNING connectors
are left alone, (3) connectors with non-STARTED target state are skipped.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java:
##########
@@ -564,7 +565,8 @@ public Set<Runnable> getCleanedStoppedTasks() {
}
public void maintainConnectorState() {
-
+ // STEP 1: redress running connectors status
Review Comment:
`redressRunningConnectors()` is called every ~1 second from
`StateMachineService` and unconditionally iterates all connectors, calling
`stateManagementService.get()` for each one. For connectors that are already in
RUNNING state (the common case), this produces unnecessary distributed status
store reads on every tick. Consider an early-exit or only checking connectors
that were recently started, similar to how the task-level redress is scoped to
`checkRunningTasks()`.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerConnector.java:
##########
@@ -369,10 +369,18 @@ public String toString() {
return sb;
}
- private enum State {
+ public enum State {
INIT,
STOPPED,
STARTED,
FAILED,
}
+
+ public State getState() {
+ return state;
+ }
+
+ public void setState(State state) {
Review Comment:
The public `setState(State)` setter is added but not used anywhere in this
PR. Exposing a public unsynchronized setter on a non-volatile field expands the
API surface and invites future callers to mutate connector state from arbitrary
threads without any synchronization guarantee. If there is no current need, it
should not be added; if needed, the field must be `volatile` and access should
be constrained.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java:
##########
@@ -564,7 +565,8 @@ public Set<Runnable> getCleanedStoppedTasks() {
}
public void maintainConnectorState() {
-
+ // STEP 1: redress running connectors status
Review Comment:
During cluster rebalancing, UNASSIGNED may be a legitimate transient state
set by the leader before reassignment completes. This redress logic could race
with the rebalance protocol by overwriting UNASSIGNED back to RUNNING on a
worker that is about to lose ownership. The three-way check (UNASSIGNED status
+ STARTED target + STARTED local state) provides some protection, but if the
local state transition to STOPPED hasn't propagated yet, a spurious RUNNING
status could be published.
--
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]
