Skip to content

[ISSUE #432] add redress running connectors status #433

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Slideee wants to merge 1 commit into
base: master
Choose a base branch
from
Open

[ISSUE #432] add redress running connectors status #433

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion ...ct-runtime/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/Worker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,7 @@

@RockteMQ-AI RockteMQ-AI Aug 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No test changes detected alongside source modifications. Consider adding tests to cover the changes.

import static org.apache.rocketmq.connect.runtime.connectorwrapper.status.AbstractStatus.State.PAUSED;
import static org.apache.rocketmq.connect.runtime.connectorwrapper.status.AbstractStatus.State.RUNNING;
import static org.apache.rocketmq.connect.runtime.connectorwrapper.status.AbstractStatus.State.UNASSIGNED;

/**
* A worker to schedule all connectors and tasks in a process.
Expand Down Expand Up @@ -564,7 +565,8 @@ public Set<Runnable> getCleanedStoppedTasks() {
}

public void maintainConnectorState() {

// STEP 1: redress running connectors status

@RockteMQ-AI RockteMQ-AI Aug 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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().

@RockteMQ-AI RockteMQ-AI Aug 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@RockteMQ-AI RockteMQ-AI Aug 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

redressRunningConnectors();
}

/**
Expand Down Expand Up @@ -935,6 +937,18 @@ private void redressRunningStatus(WorkerTask workerTask) {
}
}

@RockteMQ-AI RockteMQ-AI Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 void redressRunningConnectors() {
for (WorkerConnector connector : connectors.values()) {

@RockteMQ-AI RockteMQ-AI Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 && ...).

ConnectorStatus connectorStatus = stateManagementService.get(connector.getConnectorName());
if (connectorStatus != null && connectorStatus.getState() == UNASSIGNED && connector.getKeyValue().getTargetState() == TargetState.STARTED &&

@RockteMQ-AI RockteMQ-AI Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

connector.getState() == WorkerConnector.State.STARTED) {
ConnectorStatus redressStatus = new ConnectorStatus(connector.getConnectorName(), RUNNING, workerConfig.getWorkerId(), System.currentTimeMillis());

@RockteMQ-AI RockteMQ-AI Sep 2, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

getState() reads a non-volatile, unsynchronized field that is written by the connector executor thread, while this maintenance loop runs on StateMachineService's thread. It can therefore observe a stale STARTED after shutdown has set the connector to STOPPED, and publish RUNNING over the legitimate UNASSIGNED status. Make the state access safely published (for example, make state volatile or synchronize the read/write).

log.warn("Connector {}, Old connector status is {}, new connector status {}", connector.getConnectorName(), connectorStatus, redressStatus);
stateManagementService.put(redressStatus);
}
}
}

private Map<String, List<ConnectKeyValue>> newTasks(Map<String, List<ConnectKeyValue>> taskConfigs) {
Map<String, List<ConnectKeyValue>> newTasks = new HashMap<>();
for (String connectorName : taskConfigs.keySet()) {
Expand Down
10 changes: 9 additions & 1 deletion ...e/src/main/java/org/apache/rocketmq/connect/runtime/connectorwrapper/WorkerConnector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -369,10 +369,18 @@ public String toString() {
return sb;
}

private enum State {
public enum State {

@RockteMQ-AI RockteMQ-AI Aug 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@RockteMQ-AI RockteMQ-AI Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

INIT,
STOPPED,
STARTED,
FAILED,
}

@RockteMQ-AI RockteMQ-AI Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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;
}

@RockteMQ-AI RockteMQ-AI Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

public void setState(State state) {

@RockteMQ-AI RockteMQ-AI Aug 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

this.state = state;
}
}