[ISSUE #383]support failed task after connector restart by sunheyi6 · Pull Request #386 · apache/rocketmq-connect

sunheyi6 · 2022-12-03T04:19:28Z

What is the purpose of the change

close #383
support failed task after connector restart

Brief changelog

XX

Verifying this change

XXXX

Follow this checklist to help us incorporate your contribution quickly and easily. Notice, it would be helpful if you could finish the following 5 checklist(the last one is not necessary)before request the community to review your PR.

Make sure there is a Github issue filed for the change (usually before you start working on it). Trivial changes like typos do not require a Github issue. Your pull request should address just this issue, without pulling in other changes - one PR resolves one issue.
Format the pull request title like [ISSUE #123] Fix UnknownException when host config not exist. Each commit in the pull request should have a meaningful subject line and body.
Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
Write necessary unit-test(over 80% coverage) to verify your logic correction, more mock a little better when cross module dependency exist. If the new feature or significant change is committed, please remember to add integration-test in test module.
Run mvn -B clean apache-rat:check findbugs:findbugs checkstyle:checkstyle to make sure basic checks pass. Run mvn clean install -DskipITs to make sure unit-test pass. Run mvn clean test-compile failsafe:integration-test to make sure integration-test pass.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

odbozhou · 2022-12-06T07:34:30Z

please fix compile error

sunheyi6 · 2022-12-06T08:51:35Z

please fix compile error

@odbozhou ok，already fixed. Isn't there some tool that automatically recognizes this formatting problem?

odbozhou · 2022-12-07T06:23:28Z

please fix compile error

@odbozhou ok，already fixed. Isn't there some tool that automatically recognizes this formatting problem?

https://rocketmq.apache.org/zh/docs/4.x/contributionGuide/30code-guidelines

You can refer to this document, ide import checkstyle

odbozhou

org.apache.rocketmq.connect.runtime.service.ConfigManagementServiceImpl.ConfigChangeCallback#onCompletion

Is it missing the processing of restart connector and task in ConfigChangeCallback#onCompletion?

sunheyi6 · 2022-12-07T14:12:11Z

Is it missing the processing of restart connector and task in ConfigChangeCallback#onCompletion?

@odbozhou I have fixed the problem you mentioned, but how to stop the task, I still don't understand

odbozhou · 2022-12-08T05:52:53Z

Is it missing the processing of restart connector and task in ConfigChangeCallback#onCompletion?

@odbozhou I have fixed the problem you mentioned, but how to stop the task, I still don't understand

org.apache.rocketmq.connect.runtime.connectorwrapper.Worker#awaitStopTask

The worker has provided related methods to wait for the stop task

MemoryConfigManagementServiceImpl#restartTask

RockteMQ-AI · 2026-07-30T10:45:39Z

⚠️ Merge conflict detected

This PR has conflicts with the base branch and cannot be merged. Please rebase or merge the base branch into your branch and resolve the conflicts:

git fetch origin
git checkout master
git rebase origin/main
# resolve conflicts, then:
git push --force-with-lease

This is a one-time reminder. Feel free to @mention me for a re-review after conflicts are resolved.

Automated notification by github-manager-bot

RockteMQ-AI

Summary

This PR modifies 8 file(s) with 630 lines of diff. Changes look reasonable.

Automated review by github-manager-bot

Additional notes (not anchored to a changed line)

[INFO] rocketmq-connect-runtime/src/main/java/org/apache/rocketmq/connect/runtime/controller/AbstractConnectController.java:1 — Large diff (630 lines). Consider breaking into smaller, focused PRs for easier review. (line outside diff)

RockteMQ-AI

Summary

PR received and logged for review. This PR requires detailed code review by a maintainer.

Diff size: 630 lines
Author: sunheyi6 (NONE)

Automated review by RockteMQ-AI

RockteMQ-AI

Summary

Review of PR #386: [ISSUE #383]support failed task after connector restart

Findings: 10 issue(s) identified (3 critical).
CLA: unknown

Please address the inline comments above.

Automated review by github-manager-bot

RockteMQ-AI · 2026-08-23T20:39:45Z

+
+    /**
+     * process restart task
+     *


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.

RockteMQ-AI · 2026-08-23T20:39:45Z

                processDeleteConnectorRecord(connectorName, schemaAndValue);

+            } else if (key.startsWith(RESTART_CONNECTOR_PREFIX)) {
+                if (key.contains(TASK_PREFIX)) {


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.

RockteMQ-AI · 2026-08-23T20:39:45Z

+    public static final String TASK_PREFIX = "task-";
+
+    private static final String FIELD_EPOCH = "epoch";
+


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.

RockteMQ-AI · 2026-08-23T20:39:45Z

        }
    }

+    /**


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.

RockteMQ-AI · 2026-08-23T20:39:45Z

+     *
+     * @param connectorName
+     * @param schemaAndValue
+     */


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.

RockteMQ-AI · 2026-08-23T20:39:45Z

+     *
+     * @param connectorName
+     * @param taskNum
+     * @param schemaAndValue


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.

RockteMQ-AI · 2026-08-23T20:39:46Z

-            .build();
+        .field(FIELD_STATE, SchemaBuilder.string().build())
+        .field(FIELD_EPOCH, SchemaBuilder.int64().build())
+        .field(FIELD_PROPS,


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.

RockteMQ-AI · 2026-08-23T20:39:46Z


    }

+    @Override


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.

RockteMQ-AI · 2026-08-23T20:39:46Z

+
+    public void handleRestartTask(Context context) {
+        try {
+            String connectorName = context.pathParam(CONNECTOR_NAME);


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.

RockteMQ-AI · 2026-08-23T20:39:46Z

        Future<?> future = eventLoopGroup.shutdownGracefully();
        try {
-            future.get();
+            Object o = future.get();


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.

support failed task after connector restart

2147ab7

odbozhou added the enhancement New feature or request label Dec 6, 2022

Fix code formatting issues

162c8a6

odbozhou reviewed Dec 7, 2022

View reviewed changes

update method name and add code in ConfigChangeCallback#onCompletion

683e9ab

Fix code formatting issues

d78c3dc

odbozhou reviewed Dec 16, 2022

View reviewed changes

Comment thread ...va/org/apache/rocketmq/connect/runtime/service/memory/MemoryConfigManagementServiceImpl.java

Comment thread .../org/apache/rocketmq/connect/runtime/controller/distributed/TestConfigManagementService.java

update MemoryConfigManagementServiceImpl#restartConnector and

738566c

MemoryConfigManagementServiceImpl#restartTask

RockteMQ-AI reviewed Aug 5, 2026

View reviewed changes

RockteMQ-AI reviewed Aug 20, 2026

View reviewed changes

RockteMQ-AI suggested changes Aug 23, 2026

View reviewed changes

		public static final String TASK_PREFIX = "task-";

		private static final String FIELD_EPOCH = "epoch";

                       }
                   }
+                  /**

Uh oh!

Conversation

sunheyi6 commented Dec 3, 2022

What is the purpose of the change

Brief changelog

Verifying this change

Uh oh!

odbozhou commented Dec 6, 2022

Uh oh!

sunheyi6 commented Dec 6, 2022

Uh oh!

odbozhou commented Dec 7, 2022

Uh oh!

odbozhou left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

sunheyi6 commented Dec 7, 2022

Uh oh!

odbozhou commented Dec 8, 2022

Uh oh!

Uh oh!

Uh oh!

RockteMQ-AI commented Jul 30, 2026

Uh oh!

RockteMQ-AI left a comment

Choose a reason for hiding this comment

Summary

Additional notes (not anchored to a changed line)

Uh oh!

RockteMQ-AI left a comment

Choose a reason for hiding this comment

Summary

Uh oh!

RockteMQ-AI left a comment

Choose a reason for hiding this comment

Summary

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants