RockteMQ-AI commented on code in PR #433:
URL: https://github.com/apache/rocketmq-connect/pull/433#discussion_r3909551501
##########
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 &&
+ connector.getState() == WorkerConnector.State.STARTED) {
+ ConnectorStatus redressStatus = new
ConnectorStatus(connector.getConnectorName(), RUNNING,
workerConfig.getWorkerId(), System.currentTimeMillis());
Review Comment:
`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).
--
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]
