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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
