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