RockteMQ-AI commented on code in PR #433:
URL: https://github.com/apache/rocketmq-connect/pull/433#discussion_r3902646861
##########
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,
}
+
Review 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.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java:
##########
@@ -935,6 +937,18 @@ private void redressRunningStatus(WorkerTask workerTask) {
}
}
+ private void redressRunningConnectors() {
+ for (WorkerConnector connector : connectors.values()) {
Review 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 && ...)`.
##########
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:
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.
##########
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;
+ }
+
Review 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.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java:
##########
@@ -935,6 +937,18 @@ private void redressRunningStatus(WorkerTask workerTask) {
}
}
Review 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.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java:
##########
@@ -935,6 +937,18 @@ private void redressRunningStatus(WorkerTask workerTask) {
}
}
+ 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 &&
Review 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.
--
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]
