You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by "joeCarf (via GitHub)" <gi...@apache.org> on 2023/05/06 16:20:02 UTC
[GitHub] [rocketmq-connect] joeCarf opened a new pull request, #496: [ISSUE #478] Support ClickHouse source and sink
joeCarf opened a new pull request, #496:
URL: https://github.com/apache/rocketmq-connect/pull/496
## What is the purpose of the change
support ClickHouse source and sink
issue link: https://github.com/apache/rocketmq-connect/issues/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`.
- [x] Make sure there is a [Github issue](https://github.com/apache/rocketmq-connect/issues) 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.
- [x] 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.
- [x] Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
- [x] 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](https://github.com/apache/rocketmq/tree/master/test).
- [x] 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](http://www.apache.org/licenses/#clas).
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [rocketmq-connect] joeCarf commented on pull request #496: [ISSUE #478] Support ClickHouse source and sink
Posted by "joeCarf (via GitHub)" <gi...@apache.org>.
joeCarf commented on PR #496:
URL: https://github.com/apache/rocketmq-connect/pull/496#issuecomment-1537259550
@odbozhou hi, please take a look! :)
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #478] Support ClickHouse source and sink [rocketmq-connect]
Posted by "RockteMQ-AI (via GitHub)" <gi...@apache.org>.
RockteMQ-AI commented on PR #496:
URL: https://github.com/apache/rocketmq-connect/pull/496#issuecomment-5400686829
**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*
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #478] Support ClickHouse source and sink [rocketmq-connect]
Posted by "RockteMQ-AI (via GitHub)" <gi...@apache.org>.
RockteMQ-AI commented on code in PR #496:
URL: https://github.com/apache/rocketmq-connect/pull/496#discussion_r3909441097
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
+
+ private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) {
+ if (config.getUserName() != null && config.getPassWord() != null) {
+ return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord());
+ }
+ if (config.getAccessToken() != null) {
+ return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+ }
+ throw new RuntimeException("Credentials cannot be empty!");
+
+ }
+
+ public boolean ping() {
+ ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout));
+ int retryCount = 0;
+
+ while (retryCount < retry) {
+ if (clientPing.ping(server, timeout)) {
+ clientPing.close();
+ return true;
+ }
+ retryCount++;
+ LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry));
+ }
+ LOGGER.error("unable to ping to clickhouse server. ");
+ clientPing.close();
+ return false;
+ }
+
+ public ClickHouseNode getServer() {
+ return this.server;
+ }
+
+ public List<ClickHouseRecord> query(String query) {
+ return query(query, ClickHouseFormat.RowBinaryWithNamesAndTypes);
+ }
+
+ public List<ClickHouseRecord> query(String query, ClickHouseFormat clickHouseFormat) {
+ int retryCount = 0;
+ Exception ce = null;
+ while (retryCount < retry) {
+ try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ ClickHouseResponse response = client.read(server)
+ .format(clickHouseFormat)
+ .query(query)
+ .execute().get()) {
+
+ List<ClickHouseRecord> recordList = new ArrayList<>();
+ for (ClickHouseRecord r : response.records()) {
+ recordList.add(r);
+ }
+ return recordList;
+
+ } catch (Exception e) {
+ retryCount++;
+ LOGGER.warn(String.format("Query retry %d out of %d", retryCount, retry), e);
+ ce = e;
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
+
+ private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) {
+ if (config.getUserName() != null && config.getPassWord() != null) {
+ return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord());
+ }
+ if (config.getAccessToken() != null) {
+ return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+ }
+ throw new RuntimeException("Credentials cannot be empty!");
+
+ }
+
+ public boolean ping() {
+ ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout));
+ int retryCount = 0;
+
+ while (retryCount < retry) {
+ if (clientPing.ping(server, timeout)) {
+ clientPing.close();
+ return true;
+ }
+ retryCount++;
+ LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry));
+ }
+ LOGGER.error("unable to ping to clickhouse server. ");
+ clientPing.close();
+ return false;
+ }
+
+ public ClickHouseNode getServer() {
+ return this.server;
+ }
+
+ public List<ClickHouseRecord> query(String query) {
+ return query(query, ClickHouseFormat.RowBinaryWithNamesAndTypes);
+ }
+
+ public List<ClickHouseRecord> query(String query, ClickHouseFormat clickHouseFormat) {
+ int retryCount = 0;
+ Exception ce = null;
+ while (retryCount < retry) {
+ try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ ClickHouseResponse response = client.read(server)
+ .format(clickHouseFormat)
+ .query(query)
+ .execute().get()) {
+
+ List<ClickHouseRecord> recordList = new ArrayList<>();
+ for (ClickHouseRecord r : response.records()) {
+ recordList.add(r);
+ }
+ return recordList;
+
+ } catch (Exception e) {
+ retryCount++;
+ LOGGER.warn(String.format("Query retry %d out of %d", retryCount, retry), e);
+ ce = e;
+ }
+ }
+ throw new RuntimeException(ce);
+
+ }
+
+ private Connection getConnection(String url, Properties properties) throws SQLException {
+ ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties);
+ Connection conn = dataSource.getConnection(config.getUserName(), config.getPassWord());
+
+ System.out.println("Connected to: " + conn.getMetaData().getURL());
+ return conn;
+ }
+
+ private boolean insertJson(String jsonString, String table, String sql, String url) {
+
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
+
+ private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) {
+ if (config.getUserName() != null && config.getPassWord() != null) {
+ return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord());
+ }
+ if (config.getAccessToken() != null) {
+ return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+ }
+ throw new RuntimeException("Credentials cannot be empty!");
+
+ }
+
+ public boolean ping() {
+ ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout));
+ int retryCount = 0;
+
+ while (retryCount < retry) {
+ if (clientPing.ping(server, timeout)) {
+ clientPing.close();
+ return true;
+ }
+ retryCount++;
+ LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry));
+ }
+ LOGGER.error("unable to ping to clickhouse server. ");
+ clientPing.close();
+ return false;
+ }
+
+ public ClickHouseNode getServer() {
+ return this.server;
+ }
+
+ public List<ClickHouseRecord> query(String query) {
+ return query(query, ClickHouseFormat.RowBinaryWithNamesAndTypes);
+ }
+
+ public List<ClickHouseRecord> query(String query, ClickHouseFormat clickHouseFormat) {
+ int retryCount = 0;
+ Exception ce = null;
+ while (retryCount < retry) {
+ try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ ClickHouseResponse response = client.read(server)
+ .format(clickHouseFormat)
+ .query(query)
+ .execute().get()) {
+
+ List<ClickHouseRecord> recordList = new ArrayList<>();
+ for (ClickHouseRecord r : response.records()) {
+ recordList.add(r);
+ }
+ return recordList;
+
+ } catch (Exception e) {
+ retryCount++;
+ LOGGER.warn(String.format("Query retry %d out of %d", retryCount, retry), e);
+ ce = e;
+ }
+ }
+ throw new RuntimeException(ce);
+
Review Comment:
`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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseBaseConfig.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.config;
+
+import io.openmessaging.KeyValue;
+import java.lang.reflect.Method;
+
+public class ClickHouseBaseConfig {
+
+ private String clickHouseHost;
+
+ private Integer clickHousePort;
+
+ private String database;
+
+ private String userName;
+
+ private String passWord;
+
+ private String accessToken;
+
+ private String topic;
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ public String getClickHouseHost() {
+ return clickHouseHost;
+ }
+
+ public void setClickHouseHost(String clickHouseHost) {
+ this.clickHouseHost = clickHouseHost;
+ }
+
+ public Integer getClickHousePort() {
+ return clickHousePort;
+ }
+
+ public void setClickHousePort(Integer clickHousePort) {
+ this.clickHousePort = clickHousePort;
+ }
+
+ public String getUserName() {
+ return userName;
+ }
+
+ public void setUserName(String userName) {
+ this.userName = userName;
+ }
+
+ public String getPassWord() {
+ return passWord;
+ }
+
+ public void setPassWord(String passWord) {
+ this.passWord = passWord;
+ }
+
+ public String getAccessToken() {
+ return accessToken;
+ }
+
+ public void setAccessToken(String accessToken) {
+ this.accessToken = accessToken;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ public void setDatabase(String database) {
+ this.database = database;
+ }
+
+ public void load(KeyValue props) {
+ properties2Object(props, this);
+ }
+
+ private void properties2Object(final KeyValue p, final Object object) {
+
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
+
+ private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) {
+ if (config.getUserName() != null && config.getPassWord() != null) {
+ return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord());
+ }
+ if (config.getAccessToken() != null) {
+ return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+ }
+ throw new RuntimeException("Credentials cannot be empty!");
+
+ }
+
+ public boolean ping() {
+ ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout));
+ int retryCount = 0;
+
+ while (retryCount < retry) {
+ if (clientPing.ping(server, timeout)) {
+ clientPing.close();
+ return true;
+ }
+ retryCount++;
+ LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry));
+ }
+ LOGGER.error("unable to ping to clickhouse server. ");
+ clientPing.close();
+ return false;
+ }
+
+ public ClickHouseNode getServer() {
+ return this.server;
+ }
+
+ public List<ClickHouseRecord> query(String query) {
+ return query(query, ClickHouseFormat.RowBinaryWithNamesAndTypes);
+ }
+
+ public List<ClickHouseRecord> query(String query, ClickHouseFormat clickHouseFormat) {
+ int retryCount = 0;
+ Exception ce = null;
+ while (retryCount < retry) {
+ try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ ClickHouseResponse response = client.read(server)
+ .format(clickHouseFormat)
+ .query(query)
+ .execute().get()) {
+
Review Comment:
`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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.value.UnsignedByte;
+import com.clickhouse.data.value.UnsignedInteger;
+import com.clickhouse.data.value.UnsignedShort;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.source.SourceTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseSourceTask extends SourceTask {
+
+ private static final Logger log = LoggerFactory.getLogger(ClickHouseSourceTask.class);
+
+ private ClickHouseSourceConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public List<ConnectRecord> poll() {
+ List<ConnectRecord> res = new ArrayList<>();
+ long offset = readRecordOffset();
+ String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset);
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkConnector.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.sink;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.Task;
+import io.openmessaging.connector.api.component.task.sink.SinkConnector;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSinkConfig;
+
+public class ClickHouseSinkConnector extends SinkConnector {
+
+ private KeyValue keyValue;
+
+ @Override public List<KeyValue> taskConfigs(int maxTasks) {
+ List<KeyValue> configs = new ArrayList<>();
+ for (int i = 0; i < maxTasks; i++) {
+ configs.add(this.keyValue);
+ }
+ return configs;
+ }
+
+ @Override public Class<? extends Task> taskClass() {
+ return ClickHouseSinkTask.class;
+ }
Review Comment:
Same issue as the source connector: `taskConfigs` shares the same `KeyValue` reference across all tasks. Return per-task copies to prevent cross-task interference.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.value.UnsignedByte;
+import com.clickhouse.data.value.UnsignedInteger;
+import com.clickhouse.data.value.UnsignedShort;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.source.SourceTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseSourceTask extends SourceTask {
+
+ private static final Logger log = LoggerFactory.getLogger(ClickHouseSourceTask.class);
+
+ private ClickHouseSourceConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public List<ConnectRecord> poll() {
+ List<ConnectRecord> res = new ArrayList<>();
+ long offset = readRecordOffset();
+ String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset);
+
+ try {
+ List<ClickHouseRecord> recordList = helperClient.query(sql);
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceConnector.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.Task;
+import io.openmessaging.connector.api.component.task.source.SourceConnector;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+
+public class ClickHouseSourceConnector extends SourceConnector {
+
+ private KeyValue keyValue;
+
+ @Override public List<KeyValue> taskConfigs(int maxTasks) {
+ List<KeyValue> configs = new ArrayList<>();
+ for (int i = 0; i < maxTasks; i++) {
+ configs.add(this.keyValue);
+ }
+ return configs;
+ }
+
+ @Override public Class<? extends Task> taskClass() {
+ return ClickHouseSourceTask.class;
+ }
Review Comment:
`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.
##########
connectors/rocketmq-connect-clickhouse/src/test/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTaskTest.java:
##########
@@ -0,0 +1,89 @@
+package org.apache.rocketmq.connect.clickhouse.sink;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+
+
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseBaseConfig.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.config;
+
+import io.openmessaging.KeyValue;
+import java.lang.reflect.Method;
+
+public class ClickHouseBaseConfig {
+
+ private String clickHouseHost;
+
+ private Integer clickHousePort;
+
+ private String database;
+
+ private String userName;
+
+ private String passWord;
+
+ private String accessToken;
+
Review Comment:
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"`).
##########
connectors/rocketmq-connect-clickhouse/pom.xml:
##########
@@ -0,0 +1,204 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor
+ license agreements. See the NOTICE file distributed with this work for additional
+ information regarding copyright ownership. The ASF licenses this file to
+ You under the Apache License, Version 2.0 (the "License"); you may not use
+ this file except in compliance with the License. You may obtain a copy of
+ the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required
+ by applicable law or agreed to in writing, software distributed under the
+ License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS
+ OF ANY KIND, either express or implied. See the License for the specific
+ language governing permissions and limitations under the License. -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-connect-clickhouse</artifactId>
+ <version>1.0-SNAPSHOT</version>
+
+ <name>connect-clickhouse</name>
+
+ <licenses>
+ <license>
+ <name>The Apache Software License, Version 2.0</name>
+ <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+ </license>
+ </licenses>
+
+ <issueManagement>
+ <system>jira</system>
+ <url>https://issues.apache.org/jira/browse/RocketMQ</url>
+ </issueManagement>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>versions-maven-plugin</artifactId>
+ <version>2.3</version>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>clirr-maven-plugin</artifactId>
+ <version>2.7</version>
+ </plugin>
+ <plugin>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.6.1</version>
+ <configuration>
+ <source>${maven.compiler.source}</source>
+ <target>${maven.compiler.target}</target>
+ <compilerVersion>${maven.compiler.source}</compilerVersion>
+ <showDeprecation>true</showDeprecation>
+ <showWarnings>true</showWarnings>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.19.1</version>
+ <configuration>
+ <argLine>-Xms512m -Xmx1024m</argLine>
+ <forkMode>always</forkMode>
+ <includes>
+ <include>**/*Test.java</include>
+ </includes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-site-plugin</artifactId>
+ <version>3.6</version>
+ <configuration>
+ <locales>en_US</locales>
+ <outputEncoding>UTF-8</outputEncoding>
+ <inputEncoding>UTF-8</inputEncoding>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-source-plugin</artifactId>
+ <version>3.0.1</version>
+ <executions>
+ <execution>
+ <id>attach-sources</id>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <version>2.10.4</version>
+ <configuration>
+ <charset>UTF-8</charset>
+ <locale>en_US</locale>
+ <excludePackageNames>io.openmessaging.internal</excludePackageNames>
+ </configuration>
+ <executions>
+ <execution>
+ <id>aggregate</id>
+ <goals>
+ <goal>aggregate</goal>
+ </goals>
+ <phase>site</phase>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-resources-plugin</artifactId>
+ <version>3.0.2</version>
+ <configuration>
+ <encoding>${project.build.sourceEncoding}</encoding>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>findbugs-maven-plugin</artifactId>
+ <version>3.0.4</version>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <version>0.12</version>
+ <configuration>
+ <excludes>
+ <exclude>README.md</exclude>
+ <exclude>README-CN.md</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <version>3.0.0</version>
+ <configuration>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ </configuration>
+ <executions>
+ <execution>
+ <id>make-assembly</id>
+ <phase>package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+ <properties>
+ <maven.compiler.source>8</maven.compiler.source>
+ <maven.compiler.target>8</maven.compiler.target>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>io.openmessaging</groupId>
+ <artifactId>openmessaging-connector</artifactId>
+ <version>0.1.4</version>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.clickhouse</groupId>
+ <artifactId>clickhouse-jdbc</artifactId>
+ <version>0.4.5</version>
+ <!-- use uber jar with all dependencies included, change classifier to http for smaller jar -->
+ <classifier>all</classifier>
+ </dependency>
+
+ <dependency>
+ <groupId>org.lz4</groupId>
+ <artifactId>lz4-java</artifactId>
+ <version>1.8.0</version>
Review Comment:
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`).
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.value.UnsignedByte;
+import com.clickhouse.data.value.UnsignedInteger;
+import com.clickhouse.data.value.UnsignedShort;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.source.SourceTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseSourceTask extends SourceTask {
+
+ private static final Logger log = LoggerFactory.getLogger(ClickHouseSourceTask.class);
+
+ private ClickHouseSourceConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public List<ConnectRecord> poll() {
+ List<ConnectRecord> res = new ArrayList<>();
+ long offset = readRecordOffset();
+ String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset);
+
+ try {
+ List<ClickHouseRecord> recordList = helperClient.query(sql);
+ for (ClickHouseRecord clickHouseRecord : recordList) {
+ res.add(clickHouseRecord2ConnectRecord(clickHouseRecord, ++offset));
+ }
+ } catch (Exception e) {
Review Comment:
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`.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
+
+ private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) {
+ if (config.getUserName() != null && config.getPassWord() != null) {
+ return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord());
+ }
+ if (config.getAccessToken() != null) {
+ return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+ }
+ throw new RuntimeException("Credentials cannot be empty!");
+
+ }
+
+ public boolean ping() {
+ ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout));
+ int retryCount = 0;
+
+ while (retryCount < retry) {
+ if (clientPing.ping(server, timeout)) {
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/test/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTaskTest.java:
##########
@@ -0,0 +1,41 @@
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.util.List;
+import junit.framework.TestCase;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+
+import static java.lang.Thread.sleep;
+
+public class ClickHouseSourceTaskTest {
+
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.value.UnsignedByte;
+import com.clickhouse.data.value.UnsignedInteger;
+import com.clickhouse.data.value.UnsignedShort;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.source.SourceTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseSourceTask extends SourceTask {
+
+ private static final Logger log = LoggerFactory.getLogger(ClickHouseSourceTask.class);
+
+ private ClickHouseSourceConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public List<ConnectRecord> poll() {
+ List<ConnectRecord> res = new ArrayList<>();
+ long offset = readRecordOffset();
+ String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset);
+
+ try {
+ List<ClickHouseRecord> recordList = helperClient.query(sql);
+ for (ClickHouseRecord clickHouseRecord : recordList) {
+ res.add(clickHouseRecord2ConnectRecord(clickHouseRecord, ++offset));
+ }
+ } catch (Exception e) {
+ log.error(String.format("Fail to poll data from clickhouse! Table=%s offset=%d", config.getTable(), offset));
+ }
+ return res;
+ }
+
+ private long readRecordOffset() {
+ final RecordOffset positionInfo = this.sourceTaskContext.offsetStorageReader().readOffset(buildRecordPartition(config.getTable()));
+ if (positionInfo == null) {
+ return 0;
+ }
+ Object offset = positionInfo.getOffset().get(config.getTable() + "_" + ClickHouseConstants.CLICKHOUSE_OFFSET);
+ return offset == null ? 0 : Long.parseLong(offset.toString());
+ }
+
+ private String buildSql(String table, int maxNum, long offset) {
+ return String.format("SELECT * FROM `%s` LIMIT %d OFFSET %d;", table, maxNum, offset);
+ }
+
+ private ConnectRecord clickHouseRecord2ConnectRecord(ClickHouseRecord clickHouseRecord,
+ long offset) throws NoSuchFieldException, IllegalAccessException {
+ Schema schema = SchemaBuilder.struct().name(config.getTable()).build();
+ final List<Field> fields = buildFields(clickHouseRecord);
+ schema.setFields(fields);
+ final ConnectRecord connectRecord = new ConnectRecord(buildRecordPartition(config.getTable()),
+ buildRecordOffset(offset),
+ System.currentTimeMillis(),
+ schema,
+ this.buildPayLoad(fields, schema, clickHouseRecord));
Review Comment:
`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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTask.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.sink;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.sink.SinkTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.connector.api.errors.ConnectException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSinkConfig;
+
+public class ClickHouseSinkTask extends SinkTask {
+
+ public ClickHouseSinkConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public void put(List<ConnectRecord> sinkRecords) throws ConnectException {
+ if (sinkRecords == null || sinkRecords.size() < 1) {
+ return;
+ }
+ Map<String, JSONArray> valueMap = new HashMap<>();
+ for (ConnectRecord record : sinkRecords) {
+ String table = record.getSchema().getName();
+ JSONArray jsonArray = valueMap.getOrDefault(table, new JSONArray());
Review Comment:
`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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.value.UnsignedByte;
+import com.clickhouse.data.value.UnsignedInteger;
+import com.clickhouse.data.value.UnsignedShort;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.source.SourceTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseSourceTask extends SourceTask {
+
+ private static final Logger log = LoggerFactory.getLogger(ClickHouseSourceTask.class);
+
+ private ClickHouseSourceConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public List<ConnectRecord> poll() {
+ List<ConnectRecord> res = new ArrayList<>();
+ long offset = readRecordOffset();
+ String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset);
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTask.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.sink;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.sink.SinkTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.connector.api.errors.ConnectException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSinkConfig;
+
+public class ClickHouseSinkTask extends SinkTask {
+
+ public ClickHouseSinkConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public void put(List<ConnectRecord> sinkRecords) throws ConnectException {
+ if (sinkRecords == null || sinkRecords.size() < 1) {
+ return;
+ }
+ Map<String, JSONArray> valueMap = new HashMap<>();
+ for (ConnectRecord record : sinkRecords) {
+ String table = record.getSchema().getName();
+ JSONArray jsonArray = valueMap.getOrDefault(table, new JSONArray());
+
+ final List<Field> fields = record.getSchema().getFields();
+ final Struct structData = (Struct) record.getData();
+
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseConstants.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.config;
+
+public class ClickHouseConstants {
+ public static final String CLICKHOUSE_HOST = "clickhousehost";
+
+ public static final String CLICKHOUSE_PORT = "clickhouseport";
+
+ public static final String CLICKHOUSE_DATABASE = "database";
+
+ public static final String CLICKHOUSE_USERNAME = "username";
+
+ public static final String CLICKHOUSE_PASSWORD = "password";
+
+ public static final String CLICKHOUSE_ACCESSTOKEN = "accesstoken";
+
+ public static final String CLICKHOUSE_TABLE = "table";
+
+ public static final String TOPIC = "topic";
+
+ public static final String CLICKHOUSE_OFFSET = "OFFSET";
+
+ public static final String CLICKHOUSE_PARTITION = "CLICKHOUSE_PARTITION";
+
+ public static final Integer timeoutSecondsDefault = 30;
+
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/pom.xml:
##########
@@ -0,0 +1,204 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor
+ license agreements. See the NOTICE file distributed with this work for additional
+ information regarding copyright ownership. The ASF licenses this file to
+ You under the Apache License, Version 2.0 (the "License"); you may not use
+ this file except in compliance with the License. You may obtain a copy of
+ the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required
+ by applicable law or agreed to in writing, software distributed under the
+ License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS
+ OF ANY KIND, either express or implied. See the License for the specific
+ language governing permissions and limitations under the License. -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-connect-clickhouse</artifactId>
+ <version>1.0-SNAPSHOT</version>
+
+ <name>connect-clickhouse</name>
+
+ <licenses>
+ <license>
+ <name>The Apache Software License, Version 2.0</name>
+ <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+ </license>
+ </licenses>
+
+ <issueManagement>
+ <system>jira</system>
+ <url>https://issues.apache.org/jira/browse/RocketMQ</url>
+ </issueManagement>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>versions-maven-plugin</artifactId>
+ <version>2.3</version>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>clirr-maven-plugin</artifactId>
+ <version>2.7</version>
+ </plugin>
+ <plugin>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.6.1</version>
+ <configuration>
+ <source>${maven.compiler.source}</source>
+ <target>${maven.compiler.target}</target>
+ <compilerVersion>${maven.compiler.source}</compilerVersion>
+ <showDeprecation>true</showDeprecation>
+ <showWarnings>true</showWarnings>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.19.1</version>
+ <configuration>
+ <argLine>-Xms512m -Xmx1024m</argLine>
+ <forkMode>always</forkMode>
+ <includes>
+ <include>**/*Test.java</include>
+ </includes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-site-plugin</artifactId>
+ <version>3.6</version>
+ <configuration>
+ <locales>en_US</locales>
+ <outputEncoding>UTF-8</outputEncoding>
+ <inputEncoding>UTF-8</inputEncoding>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-source-plugin</artifactId>
+ <version>3.0.1</version>
+ <executions>
+ <execution>
+ <id>attach-sources</id>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <version>2.10.4</version>
+ <configuration>
+ <charset>UTF-8</charset>
+ <locale>en_US</locale>
+ <excludePackageNames>io.openmessaging.internal</excludePackageNames>
+ </configuration>
+ <executions>
+ <execution>
+ <id>aggregate</id>
+ <goals>
+ <goal>aggregate</goal>
+ </goals>
+ <phase>site</phase>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-resources-plugin</artifactId>
+ <version>3.0.2</version>
+ <configuration>
+ <encoding>${project.build.sourceEncoding}</encoding>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>findbugs-maven-plugin</artifactId>
+ <version>3.0.4</version>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <version>0.12</version>
+ <configuration>
+ <excludes>
+ <exclude>README.md</exclude>
+ <exclude>README-CN.md</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <version>3.0.0</version>
+ <configuration>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ </configuration>
+ <executions>
+ <execution>
+ <id>make-assembly</id>
+ <phase>package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+ <properties>
+ <maven.compiler.source>8</maven.compiler.source>
+ <maven.compiler.target>8</maven.compiler.target>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>io.openmessaging</groupId>
+ <artifactId>openmessaging-connector</artifactId>
+ <version>0.1.4</version>
+ <scope>compile</scope>
+ </dependency>
+ <dependency>
+ <groupId>com.clickhouse</groupId>
+ <artifactId>clickhouse-jdbc</artifactId>
+ <version>0.4.5</version>
Review Comment:
`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.
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #478] Support ClickHouse source and sink [rocketmq-connect]
Posted by "RockteMQ-AI (via GitHub)" <gi...@apache.org>.
RockteMQ-AI commented on code in PR #496:
URL: https://github.com/apache/rocketmq-connect/pull/496#discussion_r3902582853
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.value.UnsignedByte;
+import com.clickhouse.data.value.UnsignedInteger;
+import com.clickhouse.data.value.UnsignedShort;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.source.SourceTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseSourceTask extends SourceTask {
+
+ private static final Logger log = LoggerFactory.getLogger(ClickHouseSourceTask.class);
+
+ private ClickHouseSourceConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public List<ConnectRecord> poll() {
+ List<ConnectRecord> res = new ArrayList<>();
+ long offset = readRecordOffset();
+ String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset);
+
+ try {
+ List<ClickHouseRecord> recordList = helperClient.query(sql);
Review Comment:
**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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.value.UnsignedByte;
+import com.clickhouse.data.value.UnsignedInteger;
+import com.clickhouse.data.value.UnsignedShort;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.source.SourceTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseSourceTask extends SourceTask {
+
+ private static final Logger log = LoggerFactory.getLogger(ClickHouseSourceTask.class);
+
+ private ClickHouseSourceConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public List<ConnectRecord> poll() {
+ List<ConnectRecord> res = new ArrayList<>();
+ long offset = readRecordOffset();
+ String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset);
+
+ try {
+ List<ClickHouseRecord> recordList = helperClient.query(sql);
+ for (ClickHouseRecord clickHouseRecord : recordList) {
+ res.add(clickHouseRecord2ConnectRecord(clickHouseRecord, ++offset));
+ }
+ } catch (Exception e) {
+ log.error(String.format("Fail to poll data from clickhouse! Table=%s offset=%d", config.getTable(), offset));
+ }
+ return res;
+ }
+
+ private long readRecordOffset() {
+ final RecordOffset positionInfo = this.sourceTaskContext.offsetStorageReader().readOffset(buildRecordPartition(config.getTable()));
+ if (positionInfo == null) {
+ return 0;
+ }
+ Object offset = positionInfo.getOffset().get(config.getTable() + "_" + ClickHouseConstants.CLICKHOUSE_OFFSET);
+ return offset == null ? 0 : Long.parseLong(offset.toString());
+ }
+
+ private String buildSql(String table, int maxNum, long offset) {
+ return String.format("SELECT * FROM `%s` LIMIT %d OFFSET %d;", table, maxNum, offset);
+ }
+
+ private ConnectRecord clickHouseRecord2ConnectRecord(ClickHouseRecord clickHouseRecord,
+ long offset) throws NoSuchFieldException, IllegalAccessException {
+ Schema schema = SchemaBuilder.struct().name(config.getTable()).build();
+ final List<Field> fields = buildFields(clickHouseRecord);
+ schema.setFields(fields);
+ final ConnectRecord connectRecord = new ConnectRecord(buildRecordPartition(config.getTable()),
+ buildRecordOffset(offset),
+ System.currentTimeMillis(),
+ schema,
+ this.buildPayLoad(fields, schema, clickHouseRecord));
+ connectRecord.setExtensions(this.buildExtensions(clickHouseRecord));
+ return connectRecord;
+ }
+
+ private List<Field> buildFields(
+ ClickHouseRecord clickHouseRecord) throws NoSuchFieldException, IllegalAccessException {
Review Comment:
**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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
+
+ private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) {
+ if (config.getUserName() != null && config.getPassWord() != null) {
+ return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord());
+ }
+ if (config.getAccessToken() != null) {
+ return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+ }
+ throw new RuntimeException("Credentials cannot be empty!");
+
+ }
+
+ public boolean ping() {
+ ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout));
+ int retryCount = 0;
+
+ while (retryCount < retry) {
+ if (clientPing.ping(server, timeout)) {
+ clientPing.close();
+ return true;
+ }
+ retryCount++;
+ LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry));
+ }
+ LOGGER.error("unable to ping to clickhouse server. ");
+ clientPing.close();
+ return false;
+ }
+
+ public ClickHouseNode getServer() {
+ return this.server;
+ }
+
+ public List<ClickHouseRecord> query(String query) {
+ return query(query, ClickHouseFormat.RowBinaryWithNamesAndTypes);
+ }
+
+ public List<ClickHouseRecord> query(String query, ClickHouseFormat clickHouseFormat) {
+ int retryCount = 0;
+ Exception ce = null;
+ while (retryCount < retry) {
+ try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ ClickHouseResponse response = client.read(server)
+ .format(clickHouseFormat)
+ .query(query)
+ .execute().get()) {
+
+ List<ClickHouseRecord> recordList = new ArrayList<>();
+ for (ClickHouseRecord r : response.records()) {
+ recordList.add(r);
+ }
+ return recordList;
+
+ } catch (Exception e) {
+ retryCount++;
+ LOGGER.warn(String.format("Query retry %d out of %d", retryCount, retry), e);
+ ce = e;
+ }
+ }
+ throw new RuntimeException(ce);
+
+ }
+
+ private Connection getConnection(String url, Properties properties) throws SQLException {
+ ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties);
+ Connection conn = dataSource.getConnection(config.getUserName(), config.getPassWord());
+
+ System.out.println("Connected to: " + conn.getMetaData().getURL());
+ return conn;
+ }
+
+ private boolean insertJson(String jsonString, String table, String sql, String url) {
+
+ try (Connection connection = getConnection(url, new Properties());
+ PreparedStatement ps = connection.prepareStatement(sql)) {
+ ps.setObject(1, new ClickHouseWriter() {
+ @Override
+ public void write(ClickHouseOutputStream output) throws IOException {
+ output.writeBytes(jsonString.getBytes());
+ }
+ });
+ ps.executeUpdate();
+
+ } catch (Exception e) {
+ return false;
+ }
+ return true;
+ }
+
+ public void insertJson(String jsonString, String table) {
Review Comment:
**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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseBaseConfig.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.config;
+
+import io.openmessaging.KeyValue;
+import java.lang.reflect.Method;
+
+public class ClickHouseBaseConfig {
+
+ private String clickHouseHost;
+
+ private Integer clickHousePort;
+
+ private String database;
+
+ private String userName;
+
+ private String passWord;
+
+ private String accessToken;
+
+ private String topic;
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ public String getClickHouseHost() {
+ return clickHouseHost;
+ }
+
+ public void setClickHouseHost(String clickHouseHost) {
+ this.clickHouseHost = clickHouseHost;
+ }
+
+ public Integer getClickHousePort() {
+ return clickHousePort;
+ }
+
+ public void setClickHousePort(Integer clickHousePort) {
+ this.clickHousePort = clickHousePort;
+ }
+
+ public String getUserName() {
+ return userName;
+ }
+
+ public void setUserName(String userName) {
+ this.userName = userName;
+ }
+
+ public String getPassWord() {
+ return passWord;
+ }
+
+ public void setPassWord(String passWord) {
+ this.passWord = passWord;
+ }
+
+ public String getAccessToken() {
+ return accessToken;
+ }
+
+ public void setAccessToken(String accessToken) {
+ this.accessToken = accessToken;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ public void setDatabase(String database) {
+ this.database = database;
+ }
+
+ public void load(KeyValue props) {
+ properties2Object(props, this);
+ }
+
+ private void properties2Object(final KeyValue p, final Object object) {
+
+ Method[] methods = object.getClass().getMethods();
+ for (Method method : methods) {
+ String mn = method.getName();
+ if (mn.startsWith("set")) {
+ try {
+ String tmp = mn.substring(3);
+ String key = tmp.toLowerCase();
+
+ String property = p.getString(key);
+ if (property != null) {
+ Class<?>[] pt = method.getParameterTypes();
+ if (pt != null && pt.length > 0) {
+ String cn = pt[0].getSimpleName();
+ Object arg;
+ if (cn.equals("int") || cn.equals("Integer")) {
+ arg = Integer.parseInt(property);
+ } else if (cn.equals("long") || cn.equals("Long")) {
+ arg = Long.parseLong(property);
+ } else if (cn.equals("double") || cn.equals("Double")) {
+ arg = Double.parseDouble(property);
+ } else if (cn.equals("boolean") || cn.equals("Boolean")) {
+ arg = Boolean.parseBoolean(property);
+ } else if (cn.equals("float") || cn.equals("Float")) {
+ arg = Float.parseFloat(property);
+ } else if (cn.equals("String")) {
+ arg = property;
+ } else {
+ continue;
+ }
+ method.invoke(object, arg);
+ }
Review Comment:
**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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.value.UnsignedByte;
+import com.clickhouse.data.value.UnsignedInteger;
+import com.clickhouse.data.value.UnsignedShort;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.source.SourceTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseSourceTask extends SourceTask {
+
+ private static final Logger log = LoggerFactory.getLogger(ClickHouseSourceTask.class);
+
+ private ClickHouseSourceConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public List<ConnectRecord> poll() {
+ List<ConnectRecord> res = new ArrayList<>();
+ long offset = readRecordOffset();
+ String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset);
+
+ try {
+ List<ClickHouseRecord> recordList = helperClient.query(sql);
+ for (ClickHouseRecord clickHouseRecord : recordList) {
+ res.add(clickHouseRecord2ConnectRecord(clickHouseRecord, ++offset));
+ }
+ } catch (Exception e) {
+ log.error(String.format("Fail to poll data from clickhouse! Table=%s offset=%d", config.getTable(), offset));
+ }
Review Comment:
**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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTask.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.sink;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.sink.SinkTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.connector.api.errors.ConnectException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSinkConfig;
+
+public class ClickHouseSinkTask extends SinkTask {
+
+ public ClickHouseSinkConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public void put(List<ConnectRecord> sinkRecords) throws ConnectException {
+ if (sinkRecords == null || sinkRecords.size() < 1) {
+ return;
+ }
+ Map<String, JSONArray> valueMap = new HashMap<>();
+ for (ConnectRecord record : sinkRecords) {
+ String table = record.getSchema().getName();
+ JSONArray jsonArray = valueMap.getOrDefault(table, new JSONArray());
+
+ final List<Field> fields = record.getSchema().getFields();
+ final Struct structData = (Struct) record.getData();
+
+ JSONObject object = new JSONObject();
+ for (Field field : fields) {
+ object.put(field.getName(), structData.get(field));
+ }
Review Comment:
**`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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.value.UnsignedByte;
+import com.clickhouse.data.value.UnsignedInteger;
+import com.clickhouse.data.value.UnsignedShort;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.source.SourceTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseSourceTask extends SourceTask {
+
+ private static final Logger log = LoggerFactory.getLogger(ClickHouseSourceTask.class);
+
+ private ClickHouseSourceConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public List<ConnectRecord> poll() {
+ List<ConnectRecord> res = new ArrayList<>();
+ long offset = readRecordOffset();
+ String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset);
+
+ try {
+ List<ClickHouseRecord> recordList = helperClient.query(sql);
+ for (ClickHouseRecord clickHouseRecord : recordList) {
+ res.add(clickHouseRecord2ConnectRecord(clickHouseRecord, ++offset));
+ }
+ } catch (Exception e) {
+ log.error(String.format("Fail to poll data from clickhouse! Table=%s offset=%d", config.getTable(), offset));
+ }
+ return res;
+ }
+
+ private long readRecordOffset() {
+ final RecordOffset positionInfo = this.sourceTaskContext.offsetStorageReader().readOffset(buildRecordPartition(config.getTable()));
+ if (positionInfo == null) {
+ return 0;
+ }
+ Object offset = positionInfo.getOffset().get(config.getTable() + "_" + ClickHouseConstants.CLICKHOUSE_OFFSET);
Review Comment:
**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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceConnector.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.Task;
+import io.openmessaging.connector.api.component.task.source.SourceConnector;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+
+public class ClickHouseSourceConnector extends SourceConnector {
+
+ private KeyValue keyValue;
+
+ @Override public List<KeyValue> taskConfigs(int maxTasks) {
+ List<KeyValue> configs = new ArrayList<>();
+ for (int i = 0; i < maxTasks; i++) {
+ configs.add(this.keyValue);
+ }
+ return configs;
+ }
+
+ @Override public Class<? extends Task> taskClass() {
+ return ClickHouseSourceTask.class;
Review Comment:
**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.
##########
connectors/rocketmq-connect-clickhouse/src/test/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTaskTest.java:
##########
@@ -0,0 +1,89 @@
+package org.apache.rocketmq.connect.clickhouse.sink;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+
+
+class ClickHouseSinkTaskTest {
Review Comment:
**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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseConstants.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.config;
+
+public class ClickHouseConstants {
+ public static final String CLICKHOUSE_HOST = "clickhousehost";
+
+ public static final String CLICKHOUSE_PORT = "clickhouseport";
+
+ public static final String CLICKHOUSE_DATABASE = "database";
+
+ public static final String CLICKHOUSE_USERNAME = "username";
+
+ public static final String CLICKHOUSE_PASSWORD = "password";
+
+ public static final String CLICKHOUSE_ACCESSTOKEN = "accesstoken";
+
+ public static final String CLICKHOUSE_TABLE = "table";
+
+ public static final String TOPIC = "topic";
+
+ public static final String CLICKHOUSE_OFFSET = "OFFSET";
+
+ public static final String CLICKHOUSE_PARTITION = "CLICKHOUSE_PARTITION";
Review Comment:
**Constant naming violates Java conventions.** `timeoutSecondsDefault` and `retryCountDefault` should be `TIMEOUT_SECONDS_DEFAULT` and `RETRY_COUNT_DEFAULT` (UPPER_SNAKE_CASE for `static final` fields).
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.value.UnsignedByte;
+import com.clickhouse.data.value.UnsignedInteger;
+import com.clickhouse.data.value.UnsignedShort;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.source.SourceTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseSourceTask extends SourceTask {
+
+ private static final Logger log = LoggerFactory.getLogger(ClickHouseSourceTask.class);
+
+ private ClickHouseSourceConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public List<ConnectRecord> poll() {
+ List<ConnectRecord> res = new ArrayList<>();
+ long offset = readRecordOffset();
+ String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset);
+
+ try {
+ List<ClickHouseRecord> recordList = helperClient.query(sql);
+ for (ClickHouseRecord clickHouseRecord : recordList) {
+ res.add(clickHouseRecord2ConnectRecord(clickHouseRecord, ++offset));
+ }
+ } catch (Exception e) {
+ log.error(String.format("Fail to poll data from clickhouse! Table=%s offset=%d", config.getTable(), offset));
+ }
+ return res;
+ }
+
+ private long readRecordOffset() {
+ final RecordOffset positionInfo = this.sourceTaskContext.offsetStorageReader().readOffset(buildRecordPartition(config.getTable()));
+ if (positionInfo == null) {
+ return 0;
+ }
+ Object offset = positionInfo.getOffset().get(config.getTable() + "_" + ClickHouseConstants.CLICKHOUSE_OFFSET);
+ return offset == null ? 0 : Long.parseLong(offset.toString());
+ }
+
+ private String buildSql(String table, int maxNum, long offset) {
+ return String.format("SELECT * FROM `%s` LIMIT %d OFFSET %d;", table, maxNum, offset);
+ }
+
+ private ConnectRecord clickHouseRecord2ConnectRecord(ClickHouseRecord clickHouseRecord,
+ long offset) throws NoSuchFieldException, IllegalAccessException {
+ Schema schema = SchemaBuilder.struct().name(config.getTable()).build();
Review Comment:
**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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
+
+ private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) {
+ if (config.getUserName() != null && config.getPassWord() != null) {
+ return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord());
+ }
+ if (config.getAccessToken() != null) {
+ return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+ }
+ throw new RuntimeException("Credentials cannot be empty!");
+
+ }
+
+ public boolean ping() {
+ ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout));
+ int retryCount = 0;
+
+ while (retryCount < retry) {
+ if (clientPing.ping(server, timeout)) {
+ clientPing.close();
+ return true;
+ }
+ retryCount++;
+ LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry));
+ }
+ LOGGER.error("unable to ping to clickhouse server. ");
Review Comment:
**`query()` loads all records into memory.** For large result sets, collecting all `ClickHouseRecord`s into an `ArrayList` before returning can cause OOM. Consider streaming or chunked processing, especially since the source task already pages with LIMIT.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseSinkConfig.java:
##########
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.config;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class ClickHouseSinkConfig extends ClickHouseBaseConfig {
+ public static final Set<String> SINK_REQUEST_CONFIG = new HashSet<String>() {
+ {
Review Comment:
**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.
##########
connectors/rocketmq-connect-clickhouse/src/test/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTaskTest.java:
##########
@@ -0,0 +1,41 @@
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.util.List;
+import junit.framework.TestCase;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+
+import static java.lang.Thread.sleep;
+
+public class ClickHouseSourceTaskTest {
+
+// private static final String host = "120.48.26.195";
Review Comment:
**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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
+
+ private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) {
+ if (config.getUserName() != null && config.getPassWord() != null) {
+ return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord());
+ }
+ if (config.getAccessToken() != null) {
+ return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+ }
+ throw new RuntimeException("Credentials cannot be empty!");
+
+ }
+
+ public boolean ping() {
+ ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout));
+ int retryCount = 0;
+
+ while (retryCount < retry) {
+ if (clientPing.ping(server, timeout)) {
+ clientPing.close();
+ return true;
+ }
+ retryCount++;
+ LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry));
+ }
+ LOGGER.error("unable to ping to clickhouse server. ");
+ clientPing.close();
+ return false;
+ }
+
+ public ClickHouseNode getServer() {
+ return this.server;
+ }
+
+ public List<ClickHouseRecord> query(String query) {
+ return query(query, ClickHouseFormat.RowBinaryWithNamesAndTypes);
+ }
+
+ public List<ClickHouseRecord> query(String query, ClickHouseFormat clickHouseFormat) {
+ int retryCount = 0;
+ Exception ce = null;
+ while (retryCount < retry) {
+ try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ ClickHouseResponse response = client.read(server)
+ .format(clickHouseFormat)
+ .query(query)
+ .execute().get()) {
+
+ List<ClickHouseRecord> recordList = new ArrayList<>();
+ for (ClickHouseRecord r : response.records()) {
+ recordList.add(r);
+ }
+ return recordList;
+
+ } catch (Exception e) {
+ retryCount++;
+ LOGGER.warn(String.format("Query retry %d out of %d", retryCount, retry), e);
+ ce = e;
+ }
+ }
+ throw new RuntimeException(ce);
+
+ }
+
+ private Connection getConnection(String url, Properties properties) throws SQLException {
+ ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties);
+ Connection conn = dataSource.getConnection(config.getUserName(), config.getPassWord());
+
+ System.out.println("Connected to: " + conn.getMetaData().getURL());
Review Comment:
**`System.out.println` in production code.** Use the class logger (`LOGGER.info(...)`) instead of `System.out.println` for the connection message in `getConnection()`.
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #478] Support ClickHouse source and sink [rocketmq-connect]
Posted by "RockteMQ-AI (via GitHub)" <gi...@apache.org>.
RockteMQ-AI commented on code in PR #496:
URL: https://github.com/apache/rocketmq-connect/pull/496#discussion_r3721689435
##########
connectors/rocketmq-connect-clickhouse/README.md:
##########
@@ -0,0 +1,51 @@
+##### ClickHouseSourceConnector fully-qualified name
Review Comment:
Large diff (1266 lines). Consider breaking into smaller, focused PRs for easier review.
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE #478] Support ClickHouse source and sink [rocketmq-connect]
Posted by "RockteMQ-AI (via GitHub)" <gi...@apache.org>.
RockteMQ-AI commented on code in PR #496:
URL: https://github.com/apache/rocketmq-connect/pull/496#discussion_r3839482349
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
+
+ private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) {
+ if (config.getUserName() != null && config.getPassWord() != null) {
+ return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord());
+ }
+ if (config.getAccessToken() != null) {
+ return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+ }
+ throw new RuntimeException("Credentials cannot be empty!");
+
+ }
+
+ public boolean ping() {
+ ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout));
+ int retryCount = 0;
+
+ while (retryCount < retry) {
+ if (clientPing.ping(server, timeout)) {
+ clientPing.close();
+ return true;
+ }
+ retryCount++;
+ LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry));
+ }
+ LOGGER.error("unable to ping to clickhouse server. ");
+ clientPing.close();
+ return false;
+ }
+
+ public ClickHouseNode getServer() {
+ return this.server;
+ }
+
+ public List<ClickHouseRecord> query(String query) {
+ return query(query, ClickHouseFormat.RowBinaryWithNamesAndTypes);
+ }
+
+ public List<ClickHouseRecord> query(String query, ClickHouseFormat clickHouseFormat) {
+ int retryCount = 0;
+ Exception ce = null;
+ while (retryCount < retry) {
+ try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ ClickHouseResponse response = client.read(server)
+ .format(clickHouseFormat)
+ .query(query)
+ .execute().get()) {
+
+ List<ClickHouseRecord> recordList = new ArrayList<>();
+ for (ClickHouseRecord r : response.records()) {
+ recordList.add(r);
+ }
+ return recordList;
+
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
+
+ private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) {
+ if (config.getUserName() != null && config.getPassWord() != null) {
+ return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord());
+ }
+ if (config.getAccessToken() != null) {
+ return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+ }
+ throw new RuntimeException("Credentials cannot be empty!");
+
+ }
+
+ public boolean ping() {
+ ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout));
+ int retryCount = 0;
+
+ while (retryCount < retry) {
+ if (clientPing.ping(server, timeout)) {
+ clientPing.close();
+ return true;
+ }
+ retryCount++;
+ LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry));
+ }
+ LOGGER.error("unable to ping to clickhouse server. ");
+ clientPing.close();
+ return false;
+ }
+
+ public ClickHouseNode getServer() {
+ return this.server;
+ }
+
+ public List<ClickHouseRecord> query(String query) {
+ return query(query, ClickHouseFormat.RowBinaryWithNamesAndTypes);
+ }
+
+ public List<ClickHouseRecord> query(String query, ClickHouseFormat clickHouseFormat) {
+ int retryCount = 0;
+ Exception ce = null;
+ while (retryCount < retry) {
+ try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ ClickHouseResponse response = client.read(server)
+ .format(clickHouseFormat)
+ .query(query)
+ .execute().get()) {
+
+ List<ClickHouseRecord> recordList = new ArrayList<>();
+ for (ClickHouseRecord r : response.records()) {
+ recordList.add(r);
+ }
+ return recordList;
+
+ } catch (Exception e) {
+ retryCount++;
+ LOGGER.warn(String.format("Query retry %d out of %d", retryCount, retry), e);
+ ce = e;
+ }
+ }
+ throw new RuntimeException(ce);
+
+ }
+
+ private Connection getConnection(String url, Properties properties) throws SQLException {
+ ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties);
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import com.clickhouse.data.ClickHouseColumn;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.value.UnsignedByte;
+import com.clickhouse.data.value.UnsignedInteger;
+import com.clickhouse.data.value.UnsignedShort;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.source.SourceTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.RecordOffset;
+import io.openmessaging.connector.api.data.RecordPartition;
+import io.openmessaging.connector.api.data.Schema;
+import io.openmessaging.connector.api.data.SchemaBuilder;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.internal.DefaultKeyValue;
+import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseSourceTask extends SourceTask {
+
+ private static final Logger log = LoggerFactory.getLogger(ClickHouseSourceTask.class);
+
+ private ClickHouseSourceConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public List<ConnectRecord> poll() {
+ List<ConnectRecord> res = new ArrayList<>();
+ long offset = readRecordOffset();
+ String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset);
Review Comment:
The poll() method body is not shown in the diff (cut off at line 60), but based on the offset/partition constants defined (CLICKHOUSE_OFFSET, CLICKHOUSE_PARTITION), the source task must persist and restore offsets correctly to avoid re-reading or skipping rows on restart. Without seeing the implementation this cannot be confirmed, but reviewers must verify that the committed offset is actually used to resume from the correct row.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTask.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.sink;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.sink.SinkTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.connector.api.errors.ConnectException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSinkConfig;
+
+public class ClickHouseSinkTask extends SinkTask {
+
+ public ClickHouseSinkConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public void put(List<ConnectRecord> sinkRecords) throws ConnectException {
+ if (sinkRecords == null || sinkRecords.size() < 1) {
+ return;
+ }
+ Map<String, JSONArray> valueMap = new HashMap<>();
+ for (ConnectRecord record : sinkRecords) {
+ String table = record.getSchema().getName();
+ JSONArray jsonArray = valueMap.getOrDefault(table, new JSONArray());
+
+ final List<Field> fields = record.getSchema().getFields();
+ final Struct structData = (Struct) record.getData();
Review Comment:
record.getSchema().getName() is used as the target table name. If the schema name is null or does not match any ClickHouse table, the insert will fail with a cryptic error. The sink connector should require an explicit table configuration parameter or validate this mapping at startup.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseBaseConfig.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.config;
+
+import io.openmessaging.KeyValue;
+import java.lang.reflect.Method;
+
+public class ClickHouseBaseConfig {
+
+ private String clickHouseHost;
+
+ private Integer clickHousePort;
+
+ private String database;
+
+ private String userName;
+
+ private String passWord;
+
+ private String accessToken;
+
+ private String topic;
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ public String getClickHouseHost() {
+ return clickHouseHost;
+ }
+
+ public void setClickHouseHost(String clickHouseHost) {
+ this.clickHouseHost = clickHouseHost;
+ }
+
+ public Integer getClickHousePort() {
+ return clickHousePort;
+ }
+
+ public void setClickHousePort(Integer clickHousePort) {
+ this.clickHousePort = clickHousePort;
+ }
+
+ public String getUserName() {
+ return userName;
+ }
+
+ public void setUserName(String userName) {
+ this.userName = userName;
+ }
+
+ public String getPassWord() {
+ return passWord;
+ }
+
+ public void setPassWord(String passWord) {
+ this.passWord = passWord;
+ }
+
+ public String getAccessToken() {
+ return accessToken;
+ }
+
+ public void setAccessToken(String accessToken) {
+ this.accessToken = accessToken;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ public void setDatabase(String database) {
+ this.database = database;
+ }
+
+ public void load(KeyValue props) {
+ properties2Object(props, this);
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/pom.xml:
##########
@@ -0,0 +1,204 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor
+ license agreements. See the NOTICE file distributed with this work for additional
+ information regarding copyright ownership. The ASF licenses this file to
+ You under the Apache License, Version 2.0 (the "License"); you may not use
+ this file except in compliance with the License. You may obtain a copy of
+ the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required
+ by applicable law or agreed to in writing, software distributed under the
+ License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS
+ OF ANY KIND, either express or implied. See the License for the specific
+ language governing permissions and limitations under the License. -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+
+ <groupId>org.apache.rocketmq</groupId>
+ <artifactId>rocketmq-connect-clickhouse</artifactId>
+ <version>1.0-SNAPSHOT</version>
+
+ <name>connect-clickhouse</name>
+
+ <licenses>
+ <license>
+ <name>The Apache Software License, Version 2.0</name>
+ <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url>
+ </license>
+ </licenses>
+
+ <issueManagement>
+ <system>jira</system>
+ <url>https://issues.apache.org/jira/browse/RocketMQ</url>
+ </issueManagement>
+
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>versions-maven-plugin</artifactId>
+ <version>2.3</version>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>clirr-maven-plugin</artifactId>
+ <version>2.7</version>
+ </plugin>
+ <plugin>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <version>3.6.1</version>
+ <configuration>
+ <source>${maven.compiler.source}</source>
+ <target>${maven.compiler.target}</target>
+ <compilerVersion>${maven.compiler.source}</compilerVersion>
+ <showDeprecation>true</showDeprecation>
+ <showWarnings>true</showWarnings>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>2.19.1</version>
+ <configuration>
+ <argLine>-Xms512m -Xmx1024m</argLine>
+ <forkMode>always</forkMode>
+ <includes>
+ <include>**/*Test.java</include>
+ </includes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-site-plugin</artifactId>
+ <version>3.6</version>
+ <configuration>
+ <locales>en_US</locales>
+ <outputEncoding>UTF-8</outputEncoding>
+ <inputEncoding>UTF-8</inputEncoding>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-source-plugin</artifactId>
+ <version>3.0.1</version>
+ <executions>
+ <execution>
+ <id>attach-sources</id>
+ <goals>
+ <goal>jar</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-javadoc-plugin</artifactId>
+ <version>2.10.4</version>
+ <configuration>
+ <charset>UTF-8</charset>
+ <locale>en_US</locale>
+ <excludePackageNames>io.openmessaging.internal</excludePackageNames>
+ </configuration>
+ <executions>
+ <execution>
+ <id>aggregate</id>
+ <goals>
+ <goal>aggregate</goal>
+ </goals>
+ <phase>site</phase>
+ </execution>
+ </executions>
+ </plugin>
+ <plugin>
+ <artifactId>maven-resources-plugin</artifactId>
+ <version>3.0.2</version>
+ <configuration>
+ <encoding>${project.build.sourceEncoding}</encoding>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>org.codehaus.mojo</groupId>
+ <artifactId>findbugs-maven-plugin</artifactId>
+ <version>3.0.4</version>
+ </plugin>
+ <plugin>
+ <groupId>org.apache.rat</groupId>
+ <artifactId>apache-rat-plugin</artifactId>
+ <version>0.12</version>
+ <configuration>
+ <excludes>
+ <exclude>README.md</exclude>
+ <exclude>README-CN.md</exclude>
+ </excludes>
+ </configuration>
+ </plugin>
+ <plugin>
+ <artifactId>maven-assembly-plugin</artifactId>
+ <version>3.0.0</version>
+ <configuration>
+ <descriptorRefs>
+ <descriptorRef>jar-with-dependencies</descriptorRef>
+ </descriptorRefs>
+ </configuration>
+ <executions>
+ <execution>
+ <id>make-assembly</id>
+ <phase>package</phase>
+ <goals>
+ <goal>single</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+ <properties>
+ <maven.compiler.source>8</maven.compiler.source>
+ <maven.compiler.target>8</maven.compiler.target>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>io.openmessaging</groupId>
+ <artifactId>openmessaging-connector</artifactId>
+ <version>0.1.4</version>
+ <scope>compile</scope>
Review Comment:
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).
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
+
+ private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) {
+ if (config.getUserName() != null && config.getPassWord() != null) {
+ return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord());
+ }
+ if (config.getAccessToken() != null) {
+ return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+ }
+ throw new RuntimeException("Credentials cannot be empty!");
+
+ }
+
+ public boolean ping() {
+ ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout));
+ int retryCount = 0;
+
+ while (retryCount < retry) {
+ if (clientPing.ping(server, timeout)) {
+ clientPing.close();
+ return true;
+ }
+ retryCount++;
+ LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry));
+ }
+ LOGGER.error("unable to ping to clickhouse server. ");
+ clientPing.close();
+ return false;
+ }
+
+ public ClickHouseNode getServer() {
+ return this.server;
+ }
+
+ public List<ClickHouseRecord> query(String query) {
+ return query(query, ClickHouseFormat.RowBinaryWithNamesAndTypes);
+ }
+
+ public List<ClickHouseRecord> query(String query, ClickHouseFormat clickHouseFormat) {
+ int retryCount = 0;
+ Exception ce = null;
+ while (retryCount < retry) {
+ try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP);
+ ClickHouseResponse response = client.read(server)
+ .format(clickHouseFormat)
+ .query(query)
+ .execute().get()) {
+
+ List<ClickHouseRecord> recordList = new ArrayList<>();
+ for (ClickHouseRecord r : response.records()) {
+ recordList.add(r);
+ }
+ return recordList;
+
+ } catch (Exception e) {
+ retryCount++;
+ LOGGER.warn(String.format("Query retry %d out of %d", retryCount, retry), e);
+ ce = e;
+ }
+ }
+ throw new RuntimeException(ce);
+
+ }
+
+ private Connection getConnection(String url, Properties properties) throws SQLException {
+ ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties);
+ Connection conn = dataSource.getConnection(config.getUserName(), config.getPassWord());
+
+ System.out.println("Connected to: " + conn.getMetaData().getURL());
+ return conn;
+ }
+
+ private boolean insertJson(String jsonString, String table, String sql, String url) {
+
+ try (Connection connection = getConnection(url, new Properties());
+ PreparedStatement ps = connection.prepareStatement(sql)) {
+ ps.setObject(1, new ClickHouseWriter() {
+ @Override
+ public void write(ClickHouseOutputStream output) throws IOException {
+ output.writeBytes(jsonString.getBytes());
+ }
+ });
+ ps.executeUpdate();
+
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkConnector.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.sink;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.Task;
+import io.openmessaging.connector.api.component.task.sink.SinkConnector;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSinkConfig;
+
+public class ClickHouseSinkConnector extends SinkConnector {
+
+ private KeyValue keyValue;
+
+ @Override public List<KeyValue> taskConfigs(int maxTasks) {
+ List<KeyValue> configs = new ArrayList<>();
+ for (int i = 0; i < maxTasks; i++) {
+ configs.add(this.keyValue);
+ }
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
+
+ private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) {
+ if (config.getUserName() != null && config.getPassWord() != null) {
+ return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord());
+ }
+ if (config.getAccessToken() != null) {
+ return ClickHouseCredentials.fromAccessToken(config.getAccessToken());
+ }
+ throw new RuntimeException("Credentials cannot be empty!");
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseBaseConfig.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.config;
+
+import io.openmessaging.KeyValue;
+import java.lang.reflect.Method;
+
+public class ClickHouseBaseConfig {
+
+ private String clickHouseHost;
+
+ private Integer clickHousePort;
+
+ private String database;
+
+ private String userName;
+
+ private String passWord;
+
+ private String accessToken;
+
+ private String topic;
+
+ public String getTopic() {
+ return topic;
+ }
+
+ public void setTopic(String topic) {
+ this.topic = topic;
+ }
+
+ public String getClickHouseHost() {
+ return clickHouseHost;
+ }
+
+ public void setClickHouseHost(String clickHouseHost) {
+ this.clickHouseHost = clickHouseHost;
+ }
+
+ public Integer getClickHousePort() {
+ return clickHousePort;
+ }
+
+ public void setClickHousePort(Integer clickHousePort) {
+ this.clickHousePort = clickHousePort;
+ }
+
+ public String getUserName() {
+ return userName;
+ }
+
+ public void setUserName(String userName) {
+ this.userName = userName;
+ }
+
+ public String getPassWord() {
+ return passWord;
+ }
+
+ public void setPassWord(String passWord) {
+ this.passWord = passWord;
+ }
+
+ public String getAccessToken() {
+ return accessToken;
+ }
+
+ public void setAccessToken(String accessToken) {
+ this.accessToken = accessToken;
+ }
+
+ public String getDatabase() {
+ return database;
+ }
+
+ public void setDatabase(String database) {
+ this.database = database;
+ }
+
+ public void load(KeyValue props) {
+ properties2Object(props, this);
+ }
+
+ private void properties2Object(final KeyValue p, final Object object) {
+
+ Method[] methods = object.getClass().getMethods();
+ for (Method method : methods) {
+ String mn = method.getName();
+ if (mn.startsWith("set")) {
+ try {
Review Comment:
properties2Object silently ignores all exceptions via `catch (Throwable ignored)`. If a required config value fails to parse (e.g. a non-numeric string for clickhouseport), the field remains null/default and no error is surfaced, leading to a confusing NPE or connection failure later.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTask.java:
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.sink;
+
+import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.sink.SinkTask;
+import io.openmessaging.connector.api.data.ConnectRecord;
+import io.openmessaging.connector.api.data.Field;
+import io.openmessaging.connector.api.data.Struct;
+import io.openmessaging.connector.api.errors.ConnectException;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSinkConfig;
+
+public class ClickHouseSinkTask extends SinkTask {
+
+ public ClickHouseSinkConfig config;
+
+ private ClickHouseHelperClient helperClient;
+
+ @Override public void put(List<ConnectRecord> sinkRecords) throws ConnectException {
+ if (sinkRecords == null || sinkRecords.size() < 1) {
+ return;
+ }
+ Map<String, JSONArray> valueMap = new HashMap<>();
+ for (ConnectRecord record : sinkRecords) {
+ String table = record.getSchema().getName();
+ JSONArray jsonArray = valueMap.getOrDefault(table, new JSONArray());
+
+ final List<Field> fields = record.getSchema().getFields();
+ final Struct structData = (Struct) record.getData();
+
+ JSONObject object = new JSONObject();
+ for (Field field : fields) {
+ object.put(field.getName(), structData.get(field));
Review Comment:
record.getData() is cast unconditionally to Struct without any null or type check. If the connector receives a non-Struct payload (e.g. a plain String or Map from a JSON converter without schema), this will throw a ClassCastException and fail the entire batch silently because ConnectException is caught by the framework but the root cause may be swallowed.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java:
##########
@@ -0,0 +1,168 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.helper;
+
+import com.clickhouse.client.ClickHouseClient;
+import com.clickhouse.client.ClickHouseCredentials;
+import com.clickhouse.client.ClickHouseNode;
+import com.clickhouse.client.ClickHouseProtocol;
+import com.clickhouse.client.ClickHouseResponse;
+import com.clickhouse.data.ClickHouseFormat;
+import com.clickhouse.data.ClickHouseOutputStream;
+import com.clickhouse.data.ClickHouseRecord;
+import com.clickhouse.data.ClickHouseWriter;
+import com.clickhouse.jdbc.ClickHouseDataSource;
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ClickHouseHelperClient {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class);
+
+ private ClickHouseBaseConfig config;
+ private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC;
+ private ClickHouseNode server = null;
+ private int retry = ClickHouseConstants.retryCountDefault;
+
+ public ClickHouseHelperClient(ClickHouseBaseConfig config) {
+ this.config = config;
+ this.server = create(config);
+ }
+
+ private ClickHouseNode create(ClickHouseBaseConfig config) {
+ this.server = ClickHouseNode.builder()
+ .host(config.getClickHouseHost())
+ .port(ClickHouseProtocol.HTTP, config.getClickHousePort())
+ .database(config.getDatabase()).credentials(getCredentials(config))
+ .build();
+
+ return this.server;
+ }
Review Comment:
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.
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseSinkConfig.java:
##########
@@ -0,0 +1,30 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.config;
+
+import java.util.HashSet;
+import java.util.Set;
+
+public class ClickHouseSinkConfig extends ClickHouseBaseConfig {
+ public static final Set<String> SINK_REQUEST_CONFIG = new HashSet<String>() {
+ {
Review Comment:
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).
##########
connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceConnector.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.connect.clickhouse.source;
+
+import io.openmessaging.KeyValue;
+import io.openmessaging.connector.api.component.task.Task;
+import io.openmessaging.connector.api.component.task.source.SourceConnector;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig;
+
+public class ClickHouseSourceConnector extends SourceConnector {
+
+ private KeyValue keyValue;
+
+ @Override public List<KeyValue> taskConfigs(int maxTasks) {
+ List<KeyValue> configs = new ArrayList<>();
+ for (int i = 0; i < maxTasks; i++) {
+ configs.add(this.keyValue);
+ }
Review Comment:
Same shared-reference issue as ClickHouseSinkConnector.taskConfigs: all task slots receive the same KeyValue object reference.
--
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: commits-unsubscribe@rocketmq.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org