[ISSUE #478] Support ClickHouse source and sink by joeCarf · Pull Request #496 · apache/rocketmq-connect

joeCarf · 2023-05-06T16:20:00Z

What is the purpose of the change

support ClickHouse connectors (both source and sink)

issue link: #478

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.

joeCarf · 2023-05-07T01:10:36Z

@odbozhou hi, please take a look! :)

RockteMQ-AI

Summary

This PR modifies 13 file(s) with 1266 lines of diff. Changes look reasonable.

Automated review by github-manager-bot

RockteMQ-AI · 2026-08-05T15:04:37Z

@@ -0,0 +1,51 @@
+##### ClickHouseSourceConnector fully-qualified name


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

RockteMQ-AI

Summary

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

Diff size: 1266 lines
Author: joeCarf (CONTRIBUTOR)

Automated review by RockteMQ-AI

RockteMQ-AI

Summary

Review of PR #496: [ISSUE #478] Support ClickHouse source and sink

Findings: 15 issue(s) identified (4 critical).
CLA: unknown

Please address the inline comments above.

Automated review by github-manager-bot

Additional notes (not anchored to a changed line)

[CRITICAL] connectors/rocketmq-connect-clickhouse:1 — No tests are present anywhere in this PR. There are no unit or integration tests for the source task (including offset tracking), sink task, helper client retry logic, or config loading. This is a mandatory gap for a data integration connector where correctness of offset tracking and error handling is critical. (line outside diff)

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

+                }
+            });
+            ps.executeUpdate();
+


Infinite loop bug in insertJson: retryCount is never incremented. The loop while (retryCount < this.retry) will spin forever if insertJson returns false, causing the thread to hang indefinitely.

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

+    }
+
+    private Connection getConnection(String url, Properties properties) throws SQLException {
+        ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties);


insertJson silently swallows all exceptions and returns false with no logging. The caller has no way to distinguish a transient error from a permanent schema/data error, and the root cause is completely lost.

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

+                    recordList.add(r);
+                }
+                return recordList;
+


System.out.println used for connection diagnostics instead of the SLF4J logger already present in this class. This will not be suppressible via logging configuration and pollutes stdout in production.

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

+        if (config.getAccessToken() != null) {
+            return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+        }
+        throw new RuntimeException("Credentials cannot be empty!");


ping() creates a new ClickHouseClient on every call but only closes it on success. If ping returns false after all retries, the client opened in the last failed iteration is never closed, leaking a connection resource.

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

+            .build();
+
+        return this.server;
+    }


getCredentials throws a RuntimeException when neither username/password nor accessToken is set, but this is thrown inside the constructor during create(). The error message is not descriptive enough to guide operators. More importantly, a missing-credential config should be caught at connector validation time (in start()), not deferred to client construction.

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

+    }
+
+    public void load(KeyValue props) {
+        properties2Object(props, this);


Config key lookup lowercases the setter name (e.g. setClickHouseHost -> clickhousehost) but the constants file defines CLICKHOUSE_HOST = "clickhousehost". This works, but the mapping is fragile: adding a setter whose lowercase name doesn't match the constant (e.g. setUserName -> username vs CLICKHOUSE_USERNAME = "username") will silently fail to populate the field. userName maps to 'username' which does match, but this convention is undocumented and error-prone.

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

+
+public class ClickHouseSinkConfig extends ClickHouseBaseConfig {
+    public static final Set<String> SINK_REQUEST_CONFIG = new HashSet<String>() {
+        {


SINK_REQUEST_CONFIG only requires clickhousehost and clickhouseport, omitting database, username, and password. A connector started without credentials will pass validation but fail at runtime in getCredentials(). Minimum required fields should include database, username, and password (or accesstoken).

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

+            <groupId>io.openmessaging</groupId>
+            <artifactId>openmessaging-connector</artifactId>
+            <version>0.1.4</version>
+            <scope>compile</scope>


junit version is set to RELEASE, which is a Maven version alias that resolves to the latest available version at build time. This makes builds non-reproducible and can introduce unexpected breaking changes. Pin to a specific version (e.g. 4.13.2).

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

+        List<KeyValue> configs = new ArrayList<>();
+        for (int i = 0; i < maxTasks; i++) {
+            configs.add(this.keyValue);
+        }


taskConfigs adds the same shared KeyValue reference to every task slot. If tasks are run concurrently and any task mutates the KeyValue object, all tasks will be affected. Each task should receive an independent copy of the configuration.

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

+        List<KeyValue> configs = new ArrayList<>();
+        for (int i = 0; i < maxTasks; i++) {
+            configs.add(this.keyValue);
+        }


Same shared-reference issue as ClickHouseSinkConnector.taskConfigs: all task slots receive the same KeyValue object reference.

RockteMQ-AI · 2026-08-24T20:06:06Z

Issue Evaluation

Category: enhancement | Status: Evaluated (PR-like submission)

This issue references #478 and proposes adding ClickHouse connectors (both source and sink).

Note: This appears to be a PR submission. If you have implementation code ready, please submit it as a pull request directly.

Feasibility: Adding ClickHouse connectors is a valuable enhancement for data integration scenarios.

Age: This issue is from May 2023. If this feature is still desired, please confirm or submit a PR.

Automated evaluation by RockteMQ-AI

RockteMQ-AI

Summary

The connector has a critical infinite-loop bug in the sink retry path, uses incorrect OFFSET-based pagination for the source, relies on fragile private-field reflection, and ships with zero executable tests (all commented out with hardcoded credentials).

Findings

[CRITICAL] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:155 — Infinite loop bug. retryCount is never incremented inside the while loop. If insertJson(...) keeps returning false, this loops forever, blocking the sink task thread permanently. Add retryCount++ after the if block.
[CRITICAL] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:63 — OFFSET-based pagination is incorrect for a source connector. SELECT * LIMIT N OFFSET M will skip or duplicate rows if data is inserted or deleted between polls. It also has O(N) cost on ClickHouse for large offsets. Use a cursor-based approach (e.g., WHERE id > lastSeenId ORDER BY id LIMIT N) for correctness and performance.
[CRITICAL] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:101 — Fragile reflection hack. Accessing the private columns field of ClickHouseRecord via getDeclaredField + setAccessible will break with any library update that renames or restructures the field. Use the public API (ClickHouseRecord.getColumns() or iterate via the record's public interface) instead.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseBaseConfig.java:131 — Silently swallowing Throwable hides configuration errors. If a setter throws (e.g., NumberFormatException from a bad port value), the field stays null and the connector fails later with an unrelated error. At minimum, log the exception so misconfigurations are diagnosable.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:69 — Exception is swallowed in poll(). The catch block logs a message but does not include the exception e as a parameter, losing the stack trace. Also, returning a partial res list after an error may cause offset gaps. Pass e to the logger: log.error("...", e) and consider re-throwing or returning an empty list.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:78 — SQL injection risk. The table parameter is interpolated directly into the SQL string via String.format. While it comes from connector config (not user input per request), a misconfigured or malicious table name like tableName; DROP TABLE ... could be dangerous. Validate or quote the identifier.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTask.java:55 — record.getData() may not be a Struct. The unchecked cast (Struct) record.getData() will throw ClassCastException if the record carries a different payload type (e.g., a Map or raw bytes). Add an instanceof check before casting.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:133 — System.out.println in production code. Use the class logger (LOGGER.info(...)) instead of System.out.println for the connection message in getConnection().
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceConnector.java:40 — All tasks share the same KeyValue reference. taskConfigs adds this.keyValue to the list maxTasks times. If any task mutates the KeyValue, all tasks are affected. Return defensive copies or create new KeyValue instances per task.
[INFO] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseSinkConfig.java:25 — Double-brace initialization creates an anonymous inner class holding a reference to the outer class (if any). For a static field this is fine in practice, but prefer Collections.unmodifiableSet(new HashSet<>(Arrays.asList(...))) or Set.of(...) for clarity and to avoid the anonymous class.
[CRITICAL] connectors/rocketmq-connect-clickhouse/src/test/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTaskTest.java:18 — All test logic is commented out. There are zero executable tests in both test files. The PR description claims unit tests were written, but nothing actually runs. This must be addressed before merge — at minimum, add mocked unit tests for put(), poll(), config loading, and the retry logic.
[CRITICAL] connectors/rocketmq-connect-clickhouse/src/test/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTaskTest.java:14 — All test logic is commented out, and tests contain hardcoded credentials (120.48.26.195, password 123456). Even though commented, these should be removed entirely. Write proper unit tests using mocks instead of integration tests against a real server.
[INFO] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseConstants.java:39 — Constant naming violates Java conventions. timeoutSecondsDefault and retryCountDefault should be TIMEOUT_SECONDS_DEFAULT and RETRY_COUNT_DEFAULT (UPPER_SNAKE_CASE for static final fields).
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:90 — query() loads all records into memory. For large result sets, collecting all ClickHouseRecords into an ArrayList before returning can cause OOM. Consider streaming or chunked processing, especially since the source task already pages with LIMIT.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:88 — New Schema built on every record. clickHouseRecord2ConnectRecord creates a new Schema and field list for every single row. For tables with stable schemas, build the schema once (e.g., on first poll or start()) and reuse it. This is a significant per-record allocation on the hot path.

Automated review by github-manager-bot

RockteMQ-AI · 2026-09-01T09:10:24Z

+        return true;
+    }
+
+    public void insertJson(String jsonString, String table) {


Infinite loop bug. retryCount is never incremented inside the while loop. If insertJson(...) keeps returning false, this loops forever, blocking the sink task thread permanently. Add retryCount++ after the if block.

RockteMQ-AI · 2026-09-01T09:10:24Z

+        String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset);
+
+        try {
+            List<ClickHouseRecord> recordList = helperClient.query(sql);


OFFSET-based pagination is incorrect for a source connector. SELECT * LIMIT N OFFSET M will skip or duplicate rows if data is inserted or deleted between polls. It also has O(N) cost on ClickHouse for large offsets. Use a cursor-based approach (e.g., WHERE id > lastSeenId ORDER BY id LIMIT N) for correctness and performance.

RockteMQ-AI · 2026-09-01T09:10:24Z

+    }
+
+    private List<Field> buildFields(
+        ClickHouseRecord clickHouseRecord) throws NoSuchFieldException, IllegalAccessException {


Fragile reflection hack. Accessing the private columns field of ClickHouseRecord via getDeclaredField + setAccessible will break with any library update that renames or restructures the field. Use the public API (ClickHouseRecord.getColumns() or iterate via the record's public interface) instead.

RockteMQ-AI · 2026-09-01T09:10:24Z

+                                continue;
+                            }
+                            method.invoke(object, arg);
+                        }


Silently swallowing Throwable hides configuration errors. If a setter throws (e.g., NumberFormatException from a bad port value), the field stays null and the connector fails later with an unrelated error. At minimum, log the exception so misconfigurations are diagnosable.

RockteMQ-AI · 2026-09-01T09:10:24Z

+            }
+        } catch (Exception e) {
+            log.error(String.format("Fail to poll data from clickhouse! Table=%s offset=%d", config.getTable(), offset));
+        }


Exception is swallowed in poll(). The catch block logs a message but does not include the exception e as a parameter, losing the stack trace. Also, returning a partial res list after an error may cause offset gaps. Pass e to the logger: log.error("...", e) and consider re-throwing or returning an empty list.

RockteMQ-AI · 2026-09-01T09:10:24Z

+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+
+
+class ClickHouseSinkTaskTest {


All test logic is commented out. There are zero executable tests in both test files. The PR description claims unit tests were written, but nothing actually runs. This must be addressed before merge — at minimum, add mocked unit tests for put(), poll(), config loading, and the retry logic.

RockteMQ-AI · 2026-09-01T09:10:24Z

+
+public class ClickHouseSourceTaskTest {
+
+//    private static final String host = "120.48.26.195";


All test logic is commented out, and tests contain hardcoded credentials (120.48.26.195, password 123456). Even though commented, these should be removed entirely. Write proper unit tests using mocks instead of integration tests against a real server.

RockteMQ-AI · 2026-09-01T09:10:24Z

+
+    public static final String CLICKHOUSE_OFFSET = "OFFSET";
+
+    public static final String CLICKHOUSE_PARTITION = "CLICKHOUSE_PARTITION";


Constant naming violates Java conventions. timeoutSecondsDefault and retryCountDefault should be TIMEOUT_SECONDS_DEFAULT and RETRY_COUNT_DEFAULT (UPPER_SNAKE_CASE for static final fields).

RockteMQ-AI · 2026-09-01T09:10:24Z

+            retryCount++;
+            LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry));
+        }
+        LOGGER.error("unable to ping to clickhouse server. ");


query() loads all records into memory. For large result sets, collecting all ClickHouseRecords into an ArrayList before returning can cause OOM. Consider streaming or chunked processing, especially since the source task already pages with LIMIT.

RockteMQ-AI · 2026-09-01T09:10:24Z

+
+    private ConnectRecord clickHouseRecord2ConnectRecord(ClickHouseRecord clickHouseRecord,
+        long offset) throws NoSuchFieldException, IllegalAccessException {
+        Schema schema = SchemaBuilder.struct().name(config.getTable()).build();


New Schema built on every record. clickHouseRecord2ConnectRecord creates a new Schema and field list for every single row. For tables with stable schemas, build the schema once (e.g., on first poll or start()) and reuse it. This is a significant per-record allocation on the hot path.

RockteMQ-AI

Summary

21 finding(s) to address.

Findings

[CRITICAL] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:138 — Infinite retry loop: retryCount is never incremented inside the while loop in insertJson(String, String). If insertJson (the private overload) keeps returning false, this loop will spin forever, blocking the sink task thread indefinitely. Add retryCount++ inside the loop.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:122 — Using System.out.println for logging connection info in getConnection. Use the SLF4J logger instead to respect log-level configuration and avoid stdout pollution in production.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:126 — insertJson(String, String, String, String) swallows all exceptions silently and returns false. The caller (the public insertJson) then either retries infinitely (due to the missing increment bug) or logs only a generic error. The root-cause exception should be logged with the stack trace so failures can be diagnosed.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:112 — properties2Object in ClickHouseBaseConfig catches Throwable and silently ignores it (catch (Throwable ignored)). This masks configuration errors — e.g., a malformed port number will be silently skipped, leaving the field null and causing a confusing NullPointerException later. At minimum, log a warning.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseBaseConfig.java:100 — The reflection-based properties2Object derives config keys by stripping the set prefix and lowercasing. This means setClickHouseHost maps to key clickhousehost, setClickHousePort maps to clickhouseport, etc. While this matches the current constants, it is fragile — any rename of a setter silently changes the expected config key. A dedicated config-key-to-setter mapping or explicit annotations would be safer.
[CRITICAL] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:60 — Offset semantics are wrong: the source uses LIMIT N OFFSET offset which treats the offset as a row-number skip. This is not safe for tables with concurrent inserts/deletes — rows can be skipped or duplicated between polls. A robust approach would use an auto-incrementing primary key or a timestamp watermark (e.g., WHERE id > offset ORDER BY id LIMIT N). The current design only works correctly for static, append-only tables.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:67 — The exception in poll() is caught and logged but the error message itself does not include the exception (e) as a parameter. The stack trace is lost, making it very hard to diagnose query failures (e.g., SQL syntax errors, permission issues). Add e as a second argument to log.error.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:95 — buildFields uses reflection to access the private columns field on ClickHouseRecord (getDeclaredField("columns")). This is an internal implementation detail of the ClickHouse client library and can break with any library upgrade. Consider using the public API (e.g., clickHouseRecord.getColumns() or iterating via the public interface) if available.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:63 — The poll() method unconditionally queries SELECT * FROM table LIMIT 2000 OFFSET N. For large tables, OFFSET-based pagination becomes increasingly slow as the offset grows (ClickHouse must scan and discard all skipped rows). This is a significant performance concern for long-running connectors.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:60 — No sleep/backoff between poll() calls when data is exhausted or on error. The framework may call poll() in a tight loop, hammering the ClickHouse server with repeated identical queries. Most source connectors add a configurable poll interval or a small sleep when no new data is found.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceConnector.java:41 — taskConfigs(maxTasks) returns the same KeyValue reference for every task. If the framework or tasks mutate the config object, all tasks would see the same mutation. Consider returning defensive copies (e.g., a new DefaultKeyValue populated from the original) per task.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkConnector.java:41 — Same issue as the source connector: taskConfigs shares the same KeyValue reference across all tasks. Return per-task copies to prevent cross-task interference.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTask.java:47 — put() uses record.getSchema().getName() as the target table name. This means the producer must set the schema name to the exact ClickHouse table name. There is no fallback or configuration for a default target table, making the sink connector's behavior opaque if the schema name is null or does not match a ClickHouse table.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTask.java:51 — The sink casts record.getData() to Struct without any type check. If the record data is a Map, a primitive, or null, this will throw a ClassCastException with no meaningful error message. Add a null/type guard.
[WARNING] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseBaseConfig.java:36 — The database field has no default value and is not in the required-config sets (REQUEST_CONFIG / SINK_REQUEST_CONFIG). If omitted, ClickHouseNode.builder().database(null) and the JDBC URL will contain a null database, causing connection failures. Either make it required or provide a sensible default (e.g., "default").
[CRITICAL] connectors/rocketmq-connect-clickhouse/src/test/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTaskTest.java:17 — The entire test class is commented out — there are zero actual unit tests. The PR checklist claims >80% test coverage, but this connector has no executable tests at all. Both test files also hardcode an IP address (120.48.26.195) and credentials, which would be a security concern even if uncommented. Proper mocked unit tests are needed.
[CRITICAL] connectors/rocketmq-connect-clickhouse/src/test/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTaskTest.java:13 — Same as the sink test: the entire test is commented out, providing zero test coverage. Additionally, it imports sleep from Thread and uses while(true) — this is a manual integration script, not a unit test.
[WARNING] connectors/rocketmq-connect-clickhouse/pom.xml:176 — JUnit dependency uses <version>RELEASE</version>, which resolves to the latest release at build time. This makes builds non-reproducible and can introduce unexpected breaking changes. Pin to a specific version (e.g., 4.13.2).
[INFO] connectors/rocketmq-connect-clickhouse/pom.xml:168 — fastjson 1.2.83 is used for JSON serialization. This library has a long history of deserialization vulnerabilities. Consider using a more widely audited alternative (e.g., Jackson) which is also more commonly used in the RocketMQ ecosystem.
[INFO] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:83 — The ping() method creates a new ClickHouseClient for each call. Since ping is called during task startup, this is not a hot path, but the client should ideally be reused or at least the pattern should be consistent with the query() method which correctly uses try-with-resources.
[INFO] connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseConstants.java:42 — Naming convention inconsistency: timeoutSecondsDefault and retryCountDefault use camelCase while all other constants use UPPER_SNAKE_CASE. Also MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME uses a different naming pattern. Standardize to UPPER_SNAKE_CASE for static finals.

Automated review by github-manager-bot

RockteMQ-AI · 2026-09-01T23:43:04Z

+    }
+
+    private boolean insertJson(String jsonString, String table, String sql, String url) {
+


Infinite retry loop: retryCount is never incremented inside the while loop in insertJson(String, String). If insertJson (the private overload) keeps returning false, this loop will spin forever, blocking the sink task thread indefinitely. Add retryCount++ inside the loop.

RockteMQ-AI · 2026-09-01T23:43:04Z

+            } catch (Exception e) {
+                retryCount++;
+                LOGGER.warn(String.format("Query retry %d out of %d", retryCount, retry), e);
+                ce = e;


Using System.out.println for logging connection info in getConnection. Use the SLF4J logger instead to respect log-level configuration and avoid stdout pollution in production.

RockteMQ-AI · 2026-09-01T23:43:05Z

+            }
+        }
+        throw new RuntimeException(ce);
+


insertJson(String, String, String, String) swallows all exceptions silently and returns false. The caller (the public insertJson) then either retries infinitely (due to the missing increment bug) or logs only a generic error. The root-cause exception should be logged with the stack trace so failures can be diagnosed.

RockteMQ-AI · 2026-09-01T23:43:05Z

+                     .format(clickHouseFormat)
+                     .query(query)
+                     .execute().get()) {
+


properties2Object in ClickHouseBaseConfig catches Throwable and silently ignores it (catch (Throwable ignored)). This masks configuration errors — e.g., a malformed port number will be silently skipped, leaving the field null and causing a confusing NullPointerException later. At minimum, log a warning.

RockteMQ-AI · 2026-09-01T23:43:05Z

+    }
+
+    private void properties2Object(final KeyValue p, final Object object) {
+


The reflection-based properties2Object derives config keys by stripping the set prefix and lowercasing. This means setClickHouseHost maps to key clickhousehost, setClickHousePort maps to clickhouseport, etc. While this matches the current constants, it is fragile — any rename of a setter silently changes the expected config key. A dedicated config-key-to-setter mapping or explicit annotations would be safer.

RockteMQ-AI · 2026-09-01T23:43:05Z

+import static java.lang.Thread.sleep;
+
+public class ClickHouseSourceTaskTest {
+


Same as the sink test: the entire test is commented out, providing zero test coverage. Additionally, it imports sleep from Thread and uses while(true) — this is a manual integration script, not a unit test.

RockteMQ-AI · 2026-09-01T23:43:05Z

+        <dependency>
+            <groupId>org.lz4</groupId>
+            <artifactId>lz4-java</artifactId>
+            <version>1.8.0</version>


JUnit dependency uses <version>RELEASE</version>, which resolves to the latest release at build time. This makes builds non-reproducible and can introduce unexpected breaking changes. Pin to a specific version (e.g., 4.13.2).

RockteMQ-AI · 2026-09-01T23:43:05Z

+        <dependency>
+            <groupId>com.clickhouse</groupId>
+            <artifactId>clickhouse-jdbc</artifactId>
+            <version>0.4.5</version>


fastjson 1.2.83 is used for JSON serialization. This library has a long history of deserialization vulnerabilities. Consider using a more widely audited alternative (e.g., Jackson) which is also more commonly used in the RocketMQ ecosystem.

RockteMQ-AI · 2026-09-01T23:43:05Z

+        int retryCount = 0;
+
+        while (retryCount < retry) {
+            if (clientPing.ping(server, timeout)) {


The ping() method creates a new ClickHouseClient for each call. Since ping is called during task startup, this is not a hot path, but the client should ideally be reused or at least the pattern should be consistent with the query() method which correctly uses try-with-resources.

RockteMQ-AI · 2026-09-01T23:43:05Z

+    public static final String CLICKHOUSE_PARTITION = "CLICKHOUSE_PARTITION";
+
+    public static final Integer timeoutSecondsDefault = 30;
+


Naming convention inconsistency: timeoutSecondsDefault and retryCountDefault use camelCase while all other constants use UPPER_SNAKE_CASE. Also MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME uses a different naming pattern. Standardize to UPPER_SNAKE_CASE for static finals.

joeCarf added 8 commits May 5, 2023 17:23

initial commit

5117ede

commit

25e7ba5

rename

54c2517

finish source connector

1957ce8

finish sink connector

eb5033d

add license

969b395

Update README.md

ca0d454

Update connect-standalone.conf

31cd5e4

Merge branch 'apache:master' into dev/clickhouse-source

d5c7461

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 1, 2026

View reviewed changes

		@@ -0,0 +1,51 @@
		##### ClickHouseSourceConnector fully-qualified name

		import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;


		class ClickHouseSinkTaskTest {


		public class ClickHouseSourceTaskTest {

		// private static final String host = "120.48.26.195";


		public static final String CLICKHOUSE_OFFSET = "OFFSET";

		public static final String CLICKHOUSE_PARTITION = "CLICKHOUSE_PARTITION";

		}

		private boolean insertJson(String jsonString, String table, String sql, String url) {

		}

		private void properties2Object(final KeyValue p, final Object object) {

		import static java.lang.Thread.sleep;

		public class ClickHouseSourceTaskTest {

		public static final String CLICKHOUSE_PARTITION = "CLICKHOUSE_PARTITION";

		public static final Integer timeoutSecondsDefault = 30;

Uh oh!

Conversation

joeCarf commented May 6, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change

Brief changelog

Verifying this change

Uh oh!

joeCarf commented May 7, 2023

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 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

Additional notes (not anchored to a changed line)

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 commented Aug 24, 2026

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!

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

joeCarf commented May 6, 2023 •

edited

Loading