[ISSUE #432] add redress running connectors status - #433
Conversation
|
Can you describe what problem this PR solves? |
测试的时候发现Connector似乎也有这个问题，当发生负载均衡时，一个Running的Connector状态也会UNASSIGNED，因此想要纠正它的正确状态。 |
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
This PR modifies 2 file(s) with 65 lines of diff. No test changes detected — consider adding test coverage.
Automated review by github-manager-bot
| @@ -80,6 +80,7 @@ | |||
|
|
|||
There was a problem hiding this comment.
No test changes detected alongside source modifications. Consider adding tests to cover the changes.
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
PR received and logged for review. This PR requires detailed code review by a maintainer.
Diff size: 65 lines
Author: Slideee (CONTRIBUTOR)
Automated review by RockteMQ-AI
| } | ||
|
|
||
| private enum State { | ||
| public enum State { |
There was a problem hiding this 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.
| return state; | ||
| } | ||
|
|
||
| public void setState(State state) { |
There was a problem hiding this 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.
|
|
||
| public void maintainConnectorState() { | ||
|
|
||
| // STEP 1: redress running connectors status |
There was a problem hiding this 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().
|
|
||
| public void maintainConnectorState() { | ||
|
|
||
| // STEP 1: redress running connectors status |
There was a problem hiding this 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.
|
|
||
| public void maintainConnectorState() { | ||
|
|
||
| // STEP 1: redress running connectors status |
There was a problem hiding this 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.
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
The redress logic follows the existing pattern for tasks and addresses a real issue, but introduces thread-safety risks (non-volatile state read cross-thread, unsynchronized public setter) and lacks test coverage.
Findings
- [WARNING]
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerConnector.java:378— Thestatefield is declared as a plain (non-volatile)private State state, yet this new publicgetState()is now read from theStateMachineServicebackground thread (redressRunningConnectorsinWorker.java) while the connector's own thread writes to it (e.g., indoStart,onFailure,pause). Withoutvolatileor synchronization, the reading thread may see a stale value, causing the redress logic to either skip connectors that need correction or incorrectly "fix" connectors that have already transitioned. Consider marking the fieldvolatileor routing access through a synchronized method. - [WARNING]
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerConnector.java:382— This publicsetState()has no synchronization and no callers in this PR. TheredressRunningConnectors()method inWorker.javaonly reads state — it never writes it. Exposing an unsynchronized public setter widens the API surface and allows any caller to corrupt the connector's internal state machine without coordination with the lifecycle methods (doStart,onFailure,pause, etc.) that manage transitions undersynchronized(this). If this setter isn't needed, remove it; if it is, it should be synchronized or guarded by the existing transition protocol. - [WARNING]
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java:941— The call chainconnector.getKeyValue().getTargetState()has no null guard ongetKeyValue(). IfkeyValueis ever null (e.g., during initialization or after an error), this will throw aNullPointerExceptioninside the periodicmaintainConnectorStateloop, which could suppress further maintenance iterations. Add a null check:ConnectKeyValue kv = connector.getKeyValue(); if (kv != null && kv.getTargetState() == TargetState.STARTED && ...). - [INFO]
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java:943— UsingSystem.currentTimeMillis()as thegenerationvalue works but conflates wall-clock time with a logical generation counter. This is consistent with the existingredressRunningStatus(WorkerTask)pattern, so it's acceptable for now — but be aware that clock adjustments (NTP skew, leap seconds) could produce non-monotonic generation values. A monotonically increasing counter would be more robust. - [WARNING]
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java:938— No test coverage is included in this PR. TheredressRunningConnectors()method contains non-trivial conditional logic (three conjunctive predicates) and mutates shared status viastateManagementService.put(). A unit test should verify: (1) a connector in UNASSIGNED state with STARTED target/state gets redressed to RUNNING, (2) a connector already in RUNNING is left unchanged, (3) a connector with a non-STARTED target state is left unchanged, and (4) a nullConnectorStatusfromstateManagementService.get()is handled gracefully. - [INFO]
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerConnector.java:372— ChangingStatefromprivatetopublicexposes an internal implementation detail ofWorkerConnectorto external callers. This is a minor backward-compatibility concern: any future rename or restructure of this enum would become a public API break. Consider whether a dedicated public method likeisStarted()would suffice instead of exposing the full enum.
Automated review by github-manager-bot
| STARTED, | ||
| FAILED, | ||
| } | ||
|
|
There was a problem hiding this comment.
The state field is declared as a plain (non-volatile) private State state, yet this new public getState() is now read from the StateMachineService background thread (redressRunningConnectors in Worker.java) while the connector's own thread writes to it (e.g., in doStart, onFailure, pause). Without volatile or synchronization, the reading thread may see a stale value, causing the redress logic to either skip connectors that need correction or incorrectly "fix" connectors that have already transitioned. Consider marking the field volatile or routing access through a synchronized method.
| public State getState() { | ||
| return state; | ||
| } | ||
|
|
There was a problem hiding this comment.
This public setState() has no synchronization and no callers in this PR. The redressRunningConnectors() method in Worker.java only reads state — it never writes it. Exposing an unsynchronized public setter widens the API surface and allows any caller to corrupt the connector's internal state machine without coordination with the lifecycle methods (doStart, onFailure, pause, etc.) that manage transitions under synchronized(this). If this setter isn't needed, remove it; if it is, it should be synchronized or guarded by the existing transition protocol.
| } | ||
|
|
||
| private void redressRunningConnectors() { | ||
| for (WorkerConnector connector : connectors.values()) { |
There was a problem hiding this comment.
The call chain connector.getKeyValue().getTargetState() has no null guard on getKeyValue(). If keyValue is ever null (e.g., during initialization or after an error), this will throw a NullPointerException inside the periodic maintainConnectorState loop, which could suppress further maintenance iterations. Add a null check: ConnectKeyValue kv = connector.getKeyValue(); if (kv != null && kv.getTargetState() == TargetState.STARTED && ...).
| private void redressRunningConnectors() { | ||
| for (WorkerConnector connector : connectors.values()) { | ||
| ConnectorStatus connectorStatus = stateManagementService.get(connector.getConnectorName()); | ||
| if (connectorStatus != null && connectorStatus.getState() == UNASSIGNED && connector.getKeyValue().getTargetState() == TargetState.STARTED && |
There was a problem hiding this comment.
Using System.currentTimeMillis() as the generation value works but conflates wall-clock time with a logical generation counter. This is consistent with the existing redressRunningStatus(WorkerTask) pattern, so it's acceptable for now — but be aware that clock adjustments (NTP skew, leap seconds) could produce non-monotonic generation values. A monotonically increasing counter would be more robust.
| @@ -935,6 +937,18 @@ private void redressRunningStatus(WorkerTask workerTask) { | |||
| } | |||
| } | |||
There was a problem hiding this comment.
No test coverage is included in this PR. The redressRunningConnectors() method contains non-trivial conditional logic (three conjunctive predicates) and mutates shared status via stateManagementService.put(). A unit test should verify: (1) a connector in UNASSIGNED state with STARTED target/state gets redressed to RUNNING, (2) a connector already in RUNNING is left unchanged, (3) a connector with a non-STARTED target state is left unchanged, and (4) a null ConnectorStatus from stateManagementService.get() is handled gracefully.
| } | ||
|
|
||
| private enum State { | ||
| public enum State { |
There was a problem hiding this comment.
Changing State from private to public exposes an internal implementation detail of WorkerConnector to external callers. This is a minor backward-compatibility concern: any future rename or restructure of this enum would become a public API break. Consider whether a dedicated public method like isStarted() would suffice instead of exposing the full enum.
3 participants
What is the purpose of the change
#432
Brief changelog
XX
Verifying this change
XXXX
Follow this checklist to help us incorporate your contribution quickly and easily. Notice,
it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.[ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyleto make sure basic checks pass. Runmvn clean install -DskipITsto make sure unit-test pass. Runmvn clean test-compile failsafe:integration-testto make sure integration-test pass.