[WIP][ISSUE #22]rocketmq-connect-redis adapt to the new connect api by doubleDimple · Pull Request #31 · apache/rocketmq-connect

doubleDimple · 2022-03-25T15:04:20Z

What is the purpose of the change

XXXXX

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.

RockteMQ-AI

Summary

This PR modifies 23 file(s) with 752 lines of diff. Changes look reasonable.

Automated review by github-manager-bot

RockteMQ-AI · 2026-08-05T15:14:52Z

@@ -1,4 +1,20 @@
 <?xml version="1.0" encoding="UTF-8"?>


Large diff (752 lines). Consider breaking into smaller, focused PRs for easier review.

RockteMQ-AI · 2026-08-06T12:57:30Z

⚠️ 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 support_redis_connent
git rebase origin/develop
# 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 · 2026-08-20T00:41:58Z

This PR has been marked as [WIP] since March 2022 (over 4 years) and currently has merge conflicts.

Status check: Is the Redis connector adaptation to the new Connect API still being worked on? The implementation looks substantial (source/sink connectors, event handlers, processors, and comprehensive tests), but needs to be rebased.

If this is still relevant, please update the status and rebase. If the work has been abandoned or superseded by another effort, consider closing.

Automated review by github-manager-bot

RockteMQ-AI

Summary

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

Diff size: 752 lines
Author: doubleDimple (CONTRIBUTOR)

Automated review by RockteMQ-AI

RockteMQ-AI

Summary

Review of PR #31: [WIP][ISSUE #22]rocketmq-connect-redis adapt to the new connect api

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:53:10Z

+
+
+    public Boolean push(Map<Field, Object[]> fieldMap, EntryType entryType) {
+


RedisUpdater.push() is a stub that always returns null. RedisSinkTask.put() calls this method and checks the Boolean return value without null-safety, which will cause a NullPointerException at if (!isSuccess) on every invocation. The sink task is completely non-functional.

RockteMQ-AI · 2026-08-23T20:53:11Z

+                        Object[] value = JSONObject.parseArray((String)fieldValue).toArray();
+                        if (value.length == 2) {
+                            fieldMap.put(field, value);
+                        } else {


Unsafe cast: (String) fieldValue will throw ClassCastException if the payload value is not a String. There is no type check or null guard before the cast and the subsequent JSONObject.parseArray() call.

RockteMQ-AI · 2026-08-23T20:53:11Z

+                List<Field> fields = schema.getFields();
+                Boolean parseError = false;
+                if (!fields.isEmpty()) {
+                    for (Field field : fields) {


updater field is never initialized — it is declared but no assignment appears in start() or anywhere else. Every call to put() will throw a NullPointerException when updater.push(...) is invoked.

RockteMQ-AI · 2026-08-23T20:53:11Z

+                Schema schema = sinkDataEntry.getSchema();
+                EntryType entryType = sinkDataEntry.getEntryType();
+
+                List<Field> fields = schema.getFields();


Boolean parseError is declared with boxed Boolean (object type) instead of primitive boolean. While initialized to false, using the boxed type is unnecessary here and could theoretically cause a NullPointerException if the variable were ever left uninitialized in a refactored code path.

RockteMQ-AI · 2026-08-23T20:53:11Z

+    @Override
+    public void start(KeyValue keyValue) {
+        this.kvEntryConverter = new RedisEntryConverter();
+


e.printStackTrace() is called in start() before the structured log statement. This sends the stack trace to stderr outside the logging framework, making it invisible in log aggregators. Use only LOGGER.error(...) with the exception as the last argument.

RockteMQ-AI · 2026-08-23T20:53:11Z

+                }
+            }
+    }
+


The kvEntryConverter field is initialized in start() but never used anywhere in the class. The sink path bypasses the converter and directly parses JSON in put(), making the converter dead code and leaving the abstraction incomplete.

RockteMQ-AI · 2026-08-23T20:53:11Z

+        return config;
+    }
+
+    @Override


The eventProcessor field and its associated RedisEventHandler are started in start(), but the sink task's put() method never reads from the processor — it only writes to Redis via updater. Starting a full replication event processor (which connects to Redis as a replica) in a sink task is architecturally wrong and will create a redundant Redis replication stream.

RockteMQ-AI · 2026-08-23T20:53:11Z

+
+    @Override
+    public String verifyAndSetConfig(KeyValue config) {
+        this.keyValue = config;


taskConfigs() always returns a single-element list containing the full config regardless of the requested task count. The framework may call this with a parallelism hint; the connector ignores it and cannot scale to multiple tasks.

RockteMQ-AI · 2026-08-23T20:53:11Z

+        this.eventProcessor = eventProcessor;
+    }
+
+    public Config getConfig() {


commit() is a no-op. For at-least-once delivery guarantees the framework relies on this callback to advance committed offsets. Leaving it empty means offsets are never acknowledged, which may cause the framework to redeliver all messages on restart.

RockteMQ-AI · 2026-08-23T20:53:11Z

@@ -0,0 +1,15 @@
+package org.apache.rocketmq.connect.redis.sink;


Missing Apache License header. All other new/modified files in this PR have the ASF license header added, but RedisUpdater.java does not, which will fail the apache-rat license check added in pom.xml.

RockteMQ-AI

Summary

12 finding(s) to address.

Findings

[CRITICAL] connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSinkTask.java:48 — updater is never initialized — updater.push(...) on line 93 will always throw a NullPointerException. The RedisUpdater is a stub that returns null anyway, making the entire sink path non-functional.
[CRITICAL] connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/sink/RedisUpdater.java:12 — push() always returns null (a stub). This makes the entire sink data path a no-op. The PR is marked WIP but this class should at minimum throw UnsupportedOperationException or be clearly documented as unimplemented to avoid silent data loss.
[CRITICAL] connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSinkTask.java:89 — JSONObject.parseArray((String)fieldValue) will throw an unchecked exception if fieldValue is null or not valid JSON. No null-check or try-catch protects this call, so a single malformed record will crash the entire put() batch.
[WARNING] connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSinkTask.java:96 — When updater.push() fails or parseError is true, the record is silently dropped after logging. There is no dead-letter routing, retry, or error-reporting mechanism, which risks silent data loss in production.
[WARNING] connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSinkTask.java:121 — e.printStackTrace() is used instead of logging through SLF4J. This bypasses the configured logging infrastructure and is inappropriate for production connector code.
[WARNING] connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSinkTask.java:113 — The sink task's start() creates a DefaultRedisEventProcessor and DefaultRedisEventHandler — these are source-side replication components (they listen to Redis replication stream). This is architecturally wrong for a sink task that should be writing data into Redis.
[WARNING] connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSinkConnector.java:40 — configValid is set to true in verifyAndSetConfig but never read anywhere. adminStarted is declared but never used. These are dead fields that add confusion.
[WARNING] connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSinkConnector.java:73 — taskConfigs() always returns exactly one config. For multi-partition or scaled deployments, this prevents parallelism. The connector should accept a maxTasks parameter and distribute configs accordingly.
[WARNING] connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSinkTask.java:103 — commit() is empty — there is no offset tracking or acknowledgment logic. If the framework relies on the sink task to confirm processed offsets, this could lead to duplicate processing on restart.
[INFO] connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/sink/RedisUpdater.java:1 — Missing Apache License header. All other new files in this PR include the standard ASF license block. This will also cause apache-rat:check to fail.
[INFO] connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSinkConnector.java:54 — start() only logs a message and performs no initialization (no connection test, no admin client setup). A connector start() should validate connectivity or at least initialize shared resources.
[INFO] connectors/rocketmq-connect-redis/src/main/java/org/apache/rocketmq/connect/redis/connector/RedisSinkTask.java:80 — kvEntryConverter is initialized in start() but never used in put() or anywhere else in this class. It appears to be leftover dead code.

Automated review by github-manager-bot

RockteMQ-AI · 2026-09-02T00:41:24Z

+ * author doubleDimple
+ */
+public class RedisSinkTask extends SinkTask {
+


updater is never initialized — updater.push(...) on line 93 will always throw a NullPointerException. The RedisUpdater is a stub that returns null anyway, making the entire sink path non-functional.

RockteMQ-AI · 2026-09-02T00:41:24Z

+
+
+    public Boolean push(Map<Field, Object[]> fieldMap, EntryType entryType) {
+


push() always returns null (a stub). This makes the entire sink data path a no-op. The PR is marked WIP but this class should at minimum throw UnsupportedOperationException or be clearly documented as unimplemented to avoid silent data loss.

RockteMQ-AI · 2026-09-02T00:41:24Z

+                Boolean parseError = false;
+                if (!fields.isEmpty()) {
+                    for (Field field : fields) {
+                        Object fieldValue = payloads[field.getIndex()];


JSONObject.parseArray((String)fieldValue) will throw an unchecked exception if fieldValue is null or not valid JSON. No null-check or try-catch protects this call, so a single malformed record will crash the entire put() batch.

RockteMQ-AI · 2026-09-02T00:41:24Z

+                        } else {
+                            LOGGER.error("parseArray error, fieldValue:{}", fieldValue);
+                            parseError = true;
+                        }


When updater.push() fails or parseError is true, the record is silently dropped after logging. There is no dead-letter routing, retry, or error-reporting mechanism, which risks silent data loss in production.

RockteMQ-AI · 2026-09-02T00:41:24Z

+        this.config.load(keyValue);
+        LOGGER.info("task config msg: {}", this.config.toString());
+
+        this.eventProcessor = new DefaultRedisEventProcessor(config);


e.printStackTrace() is used instead of logging through SLF4J. This bypasses the configured logging infrastructure and is inappropriate for production connector code.

RockteMQ-AI · 2026-09-02T00:41:24Z

+
+    @Override
+    public Class<? extends Task> taskClass() {
+        return RedisSinkTask.class;


taskConfigs() always returns exactly one config. For multi-partition or scaled deployments, this prevents parallelism. The connector should accept a maxTasks parameter and distribute configs accordingly.

RockteMQ-AI · 2026-09-02T00:41:24Z

+                    Boolean isSuccess = updater.push(fieldMap, entryType);
+                    if (!isSuccess) {
+                        LOGGER.error("push data error, entryType:{}, fieldMap:{}", fieldMap, entryType);
+                    }


commit() is empty — there is no offset tracking or acknowledgment logic. If the framework relies on the sink task to confirm processed offsets, this could lead to duplicate processing on restart.

RockteMQ-AI · 2026-09-02T00:41:24Z

@@ -0,0 +1,15 @@
+package org.apache.rocketmq.connect.redis.sink;


Missing Apache License header. All other new files in this PR include the standard ASF license block. This will also cause apache-rat:check to fail.

RockteMQ-AI · 2026-09-02T00:41:24Z

+    @Override
+    public void start() {
+        LOGGER.info("the redisSinkConnector is start...");
+    }


start() only logs a message and performs no initialization (no connection test, no admin client setup). A connector start() should validate connectivity or at least initialize shared resources.

RockteMQ-AI · 2026-09-02T00:41:24Z

+             //save data from MQ to redis
+            for (SinkDataEntry sinkDataEntry : sinkDataEntries) {
+                Map<Field, Object[]> fieldMap = new HashMap<>();
+                Object[] payloads = sinkDataEntry.getPayload();


kvEntryConverter is initialized in start() but never used in put() or anywhere else in this class. It appears to be leftover dead code.

doubleDimple added 4 commits March 25, 2022 22:59

support RedisSourceConnector

8c3de96

support redis sink

9158245

support RedisSourceConnector

8359063

ADD redis mode licence

20ab758

odbozhou added the enhancement New feature or request label Apr 1, 2022

RedisSinkTask commit

f79cd4c

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

RockteMQ-AI suggested changes Sep 2, 2026

View reviewed changes



		public Boolean push(Map<Field, Object[]> fieldMap, EntryType entryType) {

		@@ -0,0 +1,15 @@
		package org.apache.rocketmq.connect.redis.sink;

Uh oh!

Conversation

doubleDimple commented Mar 25, 2022

What is the purpose of the change

Brief changelog

Verifying this change

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!

RockteMQ-AI commented Aug 6, 2026

Uh oh!

RockteMQ-AI commented Aug 20, 2026

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!

RockteMQ-AI left a comment

Choose a reason for hiding this comment

Summary

Findings

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