RockteMQ-AI commented on code in PR #386:
URL: https://github.com/apache/rocketmq-connect/pull/386#discussion_r3902668841
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImpl.java:
##########
@@ -463,6 +537,47 @@ private void processDeleteConnectorRecord(String
connectorName, SchemaAndValue s
}
}
+ /**
+ * process restarte connector
+ *
+ * @param connectorName
+ * @param schemaAndValue
+ */
+ private void processRestartConnectorRecord(String connectorName,
SchemaAndValue schemaAndValue) {
+ processDeleteConnectorRecord(connectorName, schemaAndValue);
+ processTargetStateRecord(connectorName, schemaAndValue);
+ }
+
+ /**
+ * process restart task
Review Comment:
Restart records contain only `epoch`, but `processTargetStateRecord`
requires `state`; moreover, it runs after `processDeleteConnectorRecord`
removes the saved connector/task configuration. A connector restart will
therefore fail or behave as a destructive delete. Preserve and reapply the
existing configuration and target state instead.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImpl.java:
##########
@@ -463,6 +537,47 @@ private void processDeleteConnectorRecord(String
connectorName, SchemaAndValue s
}
}
+ /**
+ * process restarte connector
+ *
+ * @param connectorName
+ * @param schemaAndValue
+ */
+ private void processRestartConnectorRecord(String connectorName,
SchemaAndValue schemaAndValue) {
+ processDeleteConnectorRecord(connectorName, schemaAndValue);
+ processTargetStateRecord(connectorName, schemaAndValue);
+ }
+
+ /**
+ * process restart task
+ *
+ * @param connectorName
+ * @param taskNum
+ * @param schemaAndValue
+ */
+ private void processRestartTaskRecord(String connectorName, String
taskNum, SchemaAndValue schemaAndValue) {
+ if (!connectorKeyValueStore.containsKey(connectorName)) {
+ return;
+ }
+ Struct value = (Struct) schemaAndValue.value();
+ Object epoch = value.get(FIELD_EPOCH);
+ // validate
+ ConnectKeyValue oldConfig = connectorKeyValueStore.get(connectorName);
+ Struct struct = (Struct) schemaAndValue.value();
+ Object targetState = struct.get(FIELD_STATE);
+ // config update
+ if ((Long) epoch > oldConfig.getEpoch()) {
+ // remove
+ connectorKeyValueStore.remove(connectorName);
Review Comment:
Task-restart records also contain only `epoch`, so reading `FIELD_STATE`
fails. Even if state were present, this path ignores `taskNum` and removes the
connector plus every task, turning a single-task restart into connector
configuration loss. Update only the addressed task.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryConfigManagementServiceImpl.java:
##########
@@ -53,7 +56,30 @@ public class MemoryConfigManagementServiceImpl extends
AbstractConfigManagementS
*/
private ConnectorConfigUpdateListener connectorConfigUpdateListener;
- public MemoryConfigManagementServiceImpl() {}
+ public static final String RESTART_CONNECTOR_PREFIX = "restart-";
+
+ public static final String TASK_PREFIX = "task-";
+
+ private static final String FIELD_EPOCH = "epoch";
+
+ /**
+ * Synchronize config with other workers.
+ */
Review Comment:
`dataSynchronizer` is newly introduced but is never initialized, and the new
`converter` field is never assigned in `initialize`. Both memory-mode restart
methods will throw a NullPointerException rather than restarting anything.
Initialize these dependencies or implement restart by updating local state and
triggering the listener.
--
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]
