RockteMQ-AI commented on code in PR #386:
URL: https://github.com/apache/rocketmq-connect/pull/386#discussion_r3909562162
##########
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:
The restart record contains only `epoch`, but `processTargetStateRecord`
requires a `state` field. After `processDeleteConnectorRecord` removes the
stored configuration, this call cannot restore it, so restarting a connector
deletes its configuration instead of restarting it.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/memory/MemoryConfigManagementServiceImpl.java:
##########
@@ -134,6 +173,34 @@ public void deleteConnectorConfig(String connectorName) {
triggerListener();
}
+ @Override
+ public void restartConnector(String connectorName) {
+ if (!connectorKeyValueStore.containsKey(connectorName)) {
+ throw new ConnectException("Connector [" + connectorName + "] does
not exist");
+ }
+ // new struct
+ Struct struct = new Struct(CONNECTOR_RESTART_CONFIGURATION_V0);
+ struct.put(FIELD_EPOCH, System.currentTimeMillis());
+
+ byte[] config = converter.fromConnectData(topic,
CONNECTOR_RESTART_CONFIGURATION_V0, struct);
+ dataSynchronizer.send(RESTART_CONNECTOR_KEY(connectorName), config);
+ }
+
Review Comment:
`converter` is newly declared in this class but is never assigned in
`initialize`; `dataSynchronizer` is likewise never initialized. Both restart
endpoints will therefore throw a NullPointerException for the memory-backed
runtime.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImpl.java:
##########
@@ -290,6 +313,45 @@ public void deleteConnectorConfig(String connectorName) {
dataSynchronizer.send(DELETE_CONNECTOR_KEY(connectorName), config);
}
+ /**
+ * restart connector config
+ *
+ * @param connectorName
+ */
+ @Override
+ public void restartConnector(String connectorName) {
+ if (!connectorKeyValueStore.containsKey(connectorName)) {
+ throw new ConnectException("Connector [" + connectorName + "] does
not exist");
+ }
+ // new struct
+ Struct struct = new Struct(CONNECTOR_RESTART_CONFIGURATION_V0);
+ struct.put(FIELD_EPOCH, System.currentTimeMillis());
+
+ byte[] config = converter.fromConnectData(topic,
CONNECTOR_RESTART_CONFIGURATION_V0, struct);
+ dataSynchronizer.send(RESTART_CONNECTOR_KEY(connectorName), config);
+ }
+
+ /**
+ * restart task config
+ *
+ * @param connectorName
+ * @param task
+ */
+ @Override
+ public void restartTask(String connectorName, Integer task) {
+ if (!connectorKeyValueStore.containsKey(connectorName)) {
+ throw new ConnectException("Connector [" + connectorName + "] does
not exist");
+ } else if (!taskKeyValueStore.containsKey(connectorName)) {
+ throw new ConnectException("Task [" + connectorName + "/" + task +
"] does not exist");
+ }
+ // new struct
Review Comment:
This only verifies that the connector has some task configuration, not that
the requested task ID exists. Requests such as `/tasks/999/restart` are
accepted for any connector with at least one task.
##########
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
Review Comment:
Task restart records also contain only `epoch`, yet this method reads
`FIELD_STATE`. Moreover, `taskNum` is never used; when the epoch condition is
met it removes the connector and every task configuration, not the requested
task.
--
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]
