RockteMQ-AI commented on code in PR #386:
URL: https://github.com/apache/rocketmq-connect/pull/386#discussion_r3839542169
##########
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:
processRestartTaskRecord reads FIELD_STATE from the struct, but
TASK_RESTART_CONFIGURATION_V0 only defines FIELD_EPOCH. struct.get(FIELD_STATE)
returns null, and TargetState.valueOf(targetState.toString()) on line ~560
throws NullPointerException. Unlike processTargetStateRecord which validates
with `instanceof String` before use, this method has no null check. The entire
task-restart path is broken and will always fail with NPE.
##########
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
}
}
+ /**
Review Comment:
processRestartConnectorRecord calls processDeleteConnectorRecord then
processTargetStateRecord. The restart schema CONNECTOR_RESTART_CONFIGURATION_V0
only has FIELD_EPOCH, but processTargetStateRecord expects FIELD_STATE (per
TARGET_STATE_V0). Since struct.get(FIELD_STATE) returns null, the `instanceof
String` check fails and the method logs an error and returns without doing
anything. Effectively, "restart connector" only deletes the connector — there
is no mechanism to re-create or re-start it with its previous config.
##########
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";
+
Review Comment:
The `dataSynchronizer` and `converter` fields are newly declared but never
initialized — the existing initialize() method body was not modified to assign
them, and MemoryConfigManagementServiceImpl has no
ConfigChangeCallback/onCompletion handler. Both restartConnector() and
restartTask() call dataSynchronizer.send() and converter.fromConnectData(),
which will throw NullPointerException. The memory implementation should restart
directly against the local stores rather than routing through a
DataSynchronizer that does not exist.
##########
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
+ */
Review Comment:
processRestartTaskRecord receives `taskNum` but never uses it. The method
removes the entire connector config and all task configs from both stores, then
triggers a global rebalance — this restarts the whole connector, not a single
task. The REST endpoint exposes per-task restart semantics that are not honored.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImpl.java:
##########
@@ -433,6 +495,18 @@ public void onCompletion(Throwable error, String key,
byte[] value) {
String connectorName =
key.substring(DELETE_CONNECTOR_PREFIX.length());
processDeleteConnectorRecord(connectorName, schemaAndValue);
+ } else if (key.startsWith(RESTART_CONNECTOR_PREFIX)) {
+ if (key.contains(TASK_PREFIX)) {
Review Comment:
The onCompletion dispatcher uses `key.contains(TASK_PREFIX)` ("task-") to
distinguish task-restart from connector-restart keys. Since all task-restart
keys start with "restart-task-", the check should be
`key.startsWith(RESTART_CONNECTOR_PREFIX + TASK_PREFIX)`. With `contains`, any
connector whose name includes "task-" (e.g. "my-task-connector") will be
misrouted to the task-restart branch. For short names like "task-x" this also
causes StringIndexOutOfBoundsException because substring(13, lastIndex) has
begin > end when the key has no task-number suffix.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/service/ConfigManagementServiceImpl.java:
##########
@@ -98,48 +107,62 @@ public static String DELETE_CONNECTOR_KEY(String
connectorName) {
* start signal
*/
public static final Schema START_SIGNAL_V0 = SchemaBuilder.struct()
- .field(START_SIGNAL, SchemaBuilder.string().build())
- .build();
+ .field(START_SIGNAL, SchemaBuilder.string().build())
+ .build();
/**
* connector configuration
*/
public static final Schema CONNECTOR_CONFIGURATION_V0 =
SchemaBuilder.struct()
- .field(FIELD_STATE, SchemaBuilder.string().build())
- .field(FIELD_EPOCH, SchemaBuilder.int64().build())
- .field(FIELD_PROPS,
- SchemaBuilder.map(
- SchemaBuilder.string().optional().build(),
- SchemaBuilder.string().optional().build()
- ).build())
- .build();
+ .field(FIELD_STATE, SchemaBuilder.string().build())
+ .field(FIELD_EPOCH, SchemaBuilder.int64().build())
+ .field(FIELD_PROPS,
Review Comment:
RESTART_TASK_KEY and RESTART_CONNECTOR_KEY can collide. For example,
RESTART_CONNECTOR_KEY("task-conn-0") produces "restart-task-conn-0", identical
to RESTART_TASK_KEY("conn", 0). The key format uses simple string concatenation
without a delimiter that cannot appear in connector names, making collisions
possible. Consider a format that unambiguously separates the namespace,
connector name, and task id.
##########
rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/rest/RestHandler.java:
##########
@@ -236,6 +240,28 @@ private void handleStopAllConnector(Context context) {
}
}
+ private void handleRestartConnector(Context context) {
+ try {
+ String connectorName = context.pathParam(CONNECTOR_NAME);
+ connectController.restartConnector(connectorName);
+ context.json(new HttpResponse<>(context.status(), "Connector [" +
connectorName + "] restarted successfully"));
+ } catch (Exception e) {
+ log.error("Restart connector failed .", e);
+ context.json(new
ErrorMessage(HttpStatus.INTERNAL_SERVER_ERROR_500, e.getMessage()));
+ }
+ }
+
+ public void handleRestartTask(Context context) {
+ try {
+ String connectorName = context.pathParam(CONNECTOR_NAME);
Review Comment:
handleRestartTask is declared public while handleRestartConnector and all
other handler methods in this class are private. This is likely unintentional
and should be private for consistency.
##########
rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/controller/distributed/TestConfigManagementService.java:
##########
@@ -58,6 +58,16 @@ public void deleteConnectorConfig(String connectorName) {
}
+ @Override
Review Comment:
No test coverage for the restart functionality. The new interface methods
restartConnector and restartTask are empty stubs in the test mock, and no unit
or integration tests exercise the restart REST endpoints, the restart record
processing, or the failure paths (nonexistent connector, invalid task id).
Given the logic bugs present, tests are essential.
##########
rocketmq-connect-runtime/src/test/java/org/apache/rocketmq/connect/runtime/connectorwrapper/ServerResponseMocker.java:
##########
@@ -98,7 +98,7 @@ public void shutdown() {
}
Future<?> future = eventLoopGroup.shutdownGracefully();
try {
- future.get();
+ Object o = future.get();
Review Comment:
The change from `future.get()` to `Object o = future.get()` introduces an
unused local variable with no functional purpose. If the goal was to suppress
an unused-return-value warning, a comment or @SuppressWarnings would be
clearer. This change appears unrelated to the PR's restart feature.
##########
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
Review Comment:
The cast `(Long) epoch` is performed without an `instanceof Long`
validation, unlike the established pattern in processTargetStateRecord and
mergeConnectConfig. If the deserialized value is not a Long (e.g., Integer from
some JSON converters, or null), this throws ClassCastException or
NullPointerException. Add an instanceof guard consistent with the rest of the
file.
--
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]
