RockteMQ-AI commented on code in PR #496: URL: https://github.com/apache/rocketmq-connect/pull/496#discussion_r3839482349
########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java: ########## @@ -0,0 +1,168 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.helper; + +import com.clickhouse.client.ClickHouseClient; +import com.clickhouse.client.ClickHouseCredentials; +import com.clickhouse.client.ClickHouseNode; +import com.clickhouse.client.ClickHouseProtocol; +import com.clickhouse.client.ClickHouseResponse; +import com.clickhouse.data.ClickHouseFormat; +import com.clickhouse.data.ClickHouseOutputStream; +import com.clickhouse.data.ClickHouseRecord; +import com.clickhouse.data.ClickHouseWriter; +import com.clickhouse.jdbc.ClickHouseDataSource; +import java.io.IOException; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ClickHouseHelperClient { + + private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class); + + private ClickHouseBaseConfig config; + private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC; + private ClickHouseNode server = null; + private int retry = ClickHouseConstants.retryCountDefault; + + public ClickHouseHelperClient(ClickHouseBaseConfig config) { + this.config = config; + this.server = create(config); + } + + private ClickHouseNode create(ClickHouseBaseConfig config) { + this.server = ClickHouseNode.builder() + .host(config.getClickHouseHost()) + .port(ClickHouseProtocol.HTTP, config.getClickHousePort()) + .database(config.getDatabase()).credentials(getCredentials(config)) + .build(); + + return this.server; + } + + private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) { + if (config.getUserName() != null && config.getPassWord() != null) { + return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord()); + } + if (config.getAccessToken() != null) { + return ClickHouseCredentials.fromAccessToken(config.getAccessToken()); + } + throw new RuntimeException("Credentials cannot be empty!"); + + } + + public boolean ping() { + ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP); + LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout)); + int retryCount = 0; + + while (retryCount < retry) { + if (clientPing.ping(server, timeout)) { + clientPing.close(); + return true; + } + retryCount++; + LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry)); + } + LOGGER.error("unable to ping to clickhouse server. "); + clientPing.close(); + return false; + } + + public ClickHouseNode getServer() { + return this.server; + } + + public List<ClickHouseRecord> query(String query) { + return query(query, ClickHouseFormat.RowBinaryWithNamesAndTypes); + } + + public List<ClickHouseRecord> query(String query, ClickHouseFormat clickHouseFormat) { + int retryCount = 0; + Exception ce = null; + while (retryCount < retry) { + try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP); + ClickHouseResponse response = client.read(server) + .format(clickHouseFormat) + .query(query) + .execute().get()) { + + List<ClickHouseRecord> recordList = new ArrayList<>(); + for (ClickHouseRecord r : response.records()) { + recordList.add(r); + } + return recordList; + Review Comment: System.out.println used for connection diagnostics instead of the SLF4J logger already present in this class. This will not be suppressible via logging configuration and pollutes stdout in production. ########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java: ########## @@ -0,0 +1,168 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.helper; + +import com.clickhouse.client.ClickHouseClient; +import com.clickhouse.client.ClickHouseCredentials; +import com.clickhouse.client.ClickHouseNode; +import com.clickhouse.client.ClickHouseProtocol; +import com.clickhouse.client.ClickHouseResponse; +import com.clickhouse.data.ClickHouseFormat; +import com.clickhouse.data.ClickHouseOutputStream; +import com.clickhouse.data.ClickHouseRecord; +import com.clickhouse.data.ClickHouseWriter; +import com.clickhouse.jdbc.ClickHouseDataSource; +import java.io.IOException; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ClickHouseHelperClient { + + private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class); + + private ClickHouseBaseConfig config; + private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC; + private ClickHouseNode server = null; + private int retry = ClickHouseConstants.retryCountDefault; + + public ClickHouseHelperClient(ClickHouseBaseConfig config) { + this.config = config; + this.server = create(config); + } + + private ClickHouseNode create(ClickHouseBaseConfig config) { + this.server = ClickHouseNode.builder() + .host(config.getClickHouseHost()) + .port(ClickHouseProtocol.HTTP, config.getClickHousePort()) + .database(config.getDatabase()).credentials(getCredentials(config)) + .build(); + + return this.server; + } + + private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) { + if (config.getUserName() != null && config.getPassWord() != null) { + return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord()); + } + if (config.getAccessToken() != null) { + return ClickHouseCredentials.fromAccessToken(config.getAccessToken()); + } + throw new RuntimeException("Credentials cannot be empty!"); + + } + + public boolean ping() { + ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP); + LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout)); + int retryCount = 0; + + while (retryCount < retry) { + if (clientPing.ping(server, timeout)) { + clientPing.close(); + return true; + } + retryCount++; + LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry)); + } + LOGGER.error("unable to ping to clickhouse server. "); + clientPing.close(); + return false; + } + + public ClickHouseNode getServer() { + return this.server; + } + + public List<ClickHouseRecord> query(String query) { + return query(query, ClickHouseFormat.RowBinaryWithNamesAndTypes); + } + + public List<ClickHouseRecord> query(String query, ClickHouseFormat clickHouseFormat) { + int retryCount = 0; + Exception ce = null; + while (retryCount < retry) { + try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP); + ClickHouseResponse response = client.read(server) + .format(clickHouseFormat) + .query(query) + .execute().get()) { + + List<ClickHouseRecord> recordList = new ArrayList<>(); + for (ClickHouseRecord r : response.records()) { + recordList.add(r); + } + return recordList; + + } catch (Exception e) { + retryCount++; + LOGGER.warn(String.format("Query retry %d out of %d", retryCount, retry), e); + ce = e; + } + } + throw new RuntimeException(ce); + + } + + private Connection getConnection(String url, Properties properties) throws SQLException { + ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties); Review Comment: insertJson silently swallows all exceptions and returns false with no logging. The caller has no way to distinguish a transient error from a permanent schema/data error, and the root cause is completely lost. ########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceTask.java: ########## @@ -0,0 +1,179 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.source; + +import com.clickhouse.data.ClickHouseColumn; +import com.clickhouse.data.ClickHouseRecord; +import com.clickhouse.data.value.UnsignedByte; +import com.clickhouse.data.value.UnsignedInteger; +import com.clickhouse.data.value.UnsignedShort; +import io.openmessaging.KeyValue; +import io.openmessaging.connector.api.component.task.source.SourceTask; +import io.openmessaging.connector.api.data.ConnectRecord; +import io.openmessaging.connector.api.data.Field; +import io.openmessaging.connector.api.data.RecordOffset; +import io.openmessaging.connector.api.data.RecordPartition; +import io.openmessaging.connector.api.data.Schema; +import io.openmessaging.connector.api.data.SchemaBuilder; +import io.openmessaging.connector.api.data.Struct; +import io.openmessaging.internal.DefaultKeyValue; +import java.sql.Timestamp; +import java.time.LocalDate; +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ClickHouseSourceTask extends SourceTask { + + private static final Logger log = LoggerFactory.getLogger(ClickHouseSourceTask.class); + + private ClickHouseSourceConfig config; + + private ClickHouseHelperClient helperClient; + + @Override public List<ConnectRecord> poll() { + List<ConnectRecord> res = new ArrayList<>(); + long offset = readRecordOffset(); + String sql = buildSql(config.getTable(), ClickHouseConstants.MAX_NUMBER_SEND_CONNECT_RECORD_EACH_TIME, offset); Review Comment: The poll() method body is not shown in the diff (cut off at line 60), but based on the offset/partition constants defined (CLICKHOUSE_OFFSET, CLICKHOUSE_PARTITION), the source task must persist and restore offsets correctly to avoid re-reading or skipping rows on restart. Without seeing the implementation this cannot be confirmed, but reviewers must verify that the committed offset is actually used to resume from the correct row. ########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTask.java: ########## @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.sink; + +import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; +import io.openmessaging.KeyValue; +import io.openmessaging.connector.api.component.task.sink.SinkTask; +import io.openmessaging.connector.api.data.ConnectRecord; +import io.openmessaging.connector.api.data.Field; +import io.openmessaging.connector.api.data.Struct; +import io.openmessaging.connector.api.errors.ConnectException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSinkConfig; + +public class ClickHouseSinkTask extends SinkTask { + + public ClickHouseSinkConfig config; + + private ClickHouseHelperClient helperClient; + + @Override public void put(List<ConnectRecord> sinkRecords) throws ConnectException { + if (sinkRecords == null || sinkRecords.size() < 1) { + return; + } + Map<String, JSONArray> valueMap = new HashMap<>(); + for (ConnectRecord record : sinkRecords) { + String table = record.getSchema().getName(); + JSONArray jsonArray = valueMap.getOrDefault(table, new JSONArray()); + + final List<Field> fields = record.getSchema().getFields(); + final Struct structData = (Struct) record.getData(); Review Comment: record.getSchema().getName() is used as the target table name. If the schema name is null or does not match any ClickHouse table, the insert will fail with a cryptic error. The sink connector should require an explicit table configuration parameter or validate this mapping at startup. ########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseBaseConfig.java: ########## @@ -0,0 +1,138 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.config; + +import io.openmessaging.KeyValue; +import java.lang.reflect.Method; + +public class ClickHouseBaseConfig { + + private String clickHouseHost; + + private Integer clickHousePort; + + private String database; + + private String userName; + + private String passWord; + + private String accessToken; + + private String topic; + + public String getTopic() { + return topic; + } + + public void setTopic(String topic) { + this.topic = topic; + } + + public String getClickHouseHost() { + return clickHouseHost; + } + + public void setClickHouseHost(String clickHouseHost) { + this.clickHouseHost = clickHouseHost; + } + + public Integer getClickHousePort() { + return clickHousePort; + } + + public void setClickHousePort(Integer clickHousePort) { + this.clickHousePort = clickHousePort; + } + + public String getUserName() { + return userName; + } + + public void setUserName(String userName) { + this.userName = userName; + } + + public String getPassWord() { + return passWord; + } + + public void setPassWord(String passWord) { + this.passWord = passWord; + } + + public String getAccessToken() { + return accessToken; + } + + public void setAccessToken(String accessToken) { + this.accessToken = accessToken; + } + + public String getDatabase() { + return database; + } + + public void setDatabase(String database) { + this.database = database; + } + + public void load(KeyValue props) { + properties2Object(props, this); Review Comment: Config key lookup lowercases the setter name (e.g. setClickHouseHost -> clickhousehost) but the constants file defines CLICKHOUSE_HOST = "clickhousehost". This works, but the mapping is fragile: adding a setter whose lowercase name doesn't match the constant (e.g. setUserName -> username vs CLICKHOUSE_USERNAME = "username") will silently fail to populate the field. userName maps to 'username' which does match, but this convention is undocumented and error-prone. ########## connectors/rocketmq-connect-clickhouse/pom.xml: ########## @@ -0,0 +1,204 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- Licensed to the Apache Software Foundation (ASF) under one or more contributor + license agreements. See the NOTICE file distributed with this work for additional + information regarding copyright ownership. The ASF licenses this file to + You under the Apache License, Version 2.0 (the "License"); you may not use + this file except in compliance with the License. You may obtain a copy of + the License at http://www.apache.org/licenses/LICENSE-2.0 Unless required + by applicable law or agreed to in writing, software distributed under the + License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS + OF ANY KIND, either express or implied. See the License for the specific + language governing permissions and limitations under the License. --> + +<project xmlns="http://maven.apache.org/POM/4.0.0"; + xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; + xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + <modelVersion>4.0.0</modelVersion> + + <groupId>org.apache.rocketmq</groupId> + <artifactId>rocketmq-connect-clickhouse</artifactId> + <version>1.0-SNAPSHOT</version> + + <name>connect-clickhouse</name> + + <licenses> + <license> + <name>The Apache Software License, Version 2.0</name> + <url>http://www.apache.org/licenses/LICENSE-2.0.txt</url> + </license> + </licenses> + + <issueManagement> + <system>jira</system> + <url>https://issues.apache.org/jira/browse/RocketMQ</url> + </issueManagement> + + <build> + <plugins> + <plugin> + <groupId>org.codehaus.mojo</groupId> + <artifactId>versions-maven-plugin</artifactId> + <version>2.3</version> + </plugin> + <plugin> + <groupId>org.codehaus.mojo</groupId> + <artifactId>clirr-maven-plugin</artifactId> + <version>2.7</version> + </plugin> + <plugin> + <artifactId>maven-compiler-plugin</artifactId> + <version>3.6.1</version> + <configuration> + <source>${maven.compiler.source}</source> + <target>${maven.compiler.target}</target> + <compilerVersion>${maven.compiler.source}</compilerVersion> + <showDeprecation>true</showDeprecation> + <showWarnings>true</showWarnings> + </configuration> + </plugin> + <plugin> + <artifactId>maven-surefire-plugin</artifactId> + <version>2.19.1</version> + <configuration> + <argLine>-Xms512m -Xmx1024m</argLine> + <forkMode>always</forkMode> + <includes> + <include>**/*Test.java</include> + </includes> + </configuration> + </plugin> + <plugin> + <artifactId>maven-site-plugin</artifactId> + <version>3.6</version> + <configuration> + <locales>en_US</locales> + <outputEncoding>UTF-8</outputEncoding> + <inputEncoding>UTF-8</inputEncoding> + </configuration> + </plugin> + <plugin> + <artifactId>maven-source-plugin</artifactId> + <version>3.0.1</version> + <executions> + <execution> + <id>attach-sources</id> + <goals> + <goal>jar</goal> + </goals> + </execution> + </executions> + </plugin> + <plugin> + <artifactId>maven-javadoc-plugin</artifactId> + <version>2.10.4</version> + <configuration> + <charset>UTF-8</charset> + <locale>en_US</locale> + <excludePackageNames>io.openmessaging.internal</excludePackageNames> + </configuration> + <executions> + <execution> + <id>aggregate</id> + <goals> + <goal>aggregate</goal> + </goals> + <phase>site</phase> + </execution> + </executions> + </plugin> + <plugin> + <artifactId>maven-resources-plugin</artifactId> + <version>3.0.2</version> + <configuration> + <encoding>${project.build.sourceEncoding}</encoding> + </configuration> + </plugin> + <plugin> + <groupId>org.codehaus.mojo</groupId> + <artifactId>findbugs-maven-plugin</artifactId> + <version>3.0.4</version> + </plugin> + <plugin> + <groupId>org.apache.rat</groupId> + <artifactId>apache-rat-plugin</artifactId> + <version>0.12</version> + <configuration> + <excludes> + <exclude>README.md</exclude> + <exclude>README-CN.md</exclude> + </excludes> + </configuration> + </plugin> + <plugin> + <artifactId>maven-assembly-plugin</artifactId> + <version>3.0.0</version> + <configuration> + <descriptorRefs> + <descriptorRef>jar-with-dependencies</descriptorRef> + </descriptorRefs> + </configuration> + <executions> + <execution> + <id>make-assembly</id> + <phase>package</phase> + <goals> + <goal>single</goal> + </goals> + </execution> + </executions> + </plugin> + </plugins> + </build> + + <properties> + <maven.compiler.source>8</maven.compiler.source> + <maven.compiler.target>8</maven.compiler.target> + <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> + </properties> + <dependencies> + <dependency> + <groupId>io.openmessaging</groupId> + <artifactId>openmessaging-connector</artifactId> + <version>0.1.4</version> + <scope>compile</scope> Review Comment: junit version is set to RELEASE, which is a Maven version alias that resolves to the latest available version at build time. This makes builds non-reproducible and can introduce unexpected breaking changes. Pin to a specific version (e.g. 4.13.2). ########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java: ########## @@ -0,0 +1,168 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.helper; + +import com.clickhouse.client.ClickHouseClient; +import com.clickhouse.client.ClickHouseCredentials; +import com.clickhouse.client.ClickHouseNode; +import com.clickhouse.client.ClickHouseProtocol; +import com.clickhouse.client.ClickHouseResponse; +import com.clickhouse.data.ClickHouseFormat; +import com.clickhouse.data.ClickHouseOutputStream; +import com.clickhouse.data.ClickHouseRecord; +import com.clickhouse.data.ClickHouseWriter; +import com.clickhouse.jdbc.ClickHouseDataSource; +import java.io.IOException; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ClickHouseHelperClient { + + private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class); + + private ClickHouseBaseConfig config; + private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC; + private ClickHouseNode server = null; + private int retry = ClickHouseConstants.retryCountDefault; + + public ClickHouseHelperClient(ClickHouseBaseConfig config) { + this.config = config; + this.server = create(config); + } + + private ClickHouseNode create(ClickHouseBaseConfig config) { + this.server = ClickHouseNode.builder() + .host(config.getClickHouseHost()) + .port(ClickHouseProtocol.HTTP, config.getClickHousePort()) + .database(config.getDatabase()).credentials(getCredentials(config)) + .build(); + + return this.server; + } + + private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) { + if (config.getUserName() != null && config.getPassWord() != null) { + return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord()); + } + if (config.getAccessToken() != null) { + return ClickHouseCredentials.fromAccessToken(config.getAccessToken()); + } + throw new RuntimeException("Credentials cannot be empty!"); + + } + + public boolean ping() { + ClickHouseClient clientPing = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP); + LOGGER.debug(String.format("server [%s] , timeout [%d]", server, timeout)); + int retryCount = 0; + + while (retryCount < retry) { + if (clientPing.ping(server, timeout)) { + clientPing.close(); + return true; + } + retryCount++; + LOGGER.warn(String.format("Ping retry %d out of %d", retryCount, retry)); + } + LOGGER.error("unable to ping to clickhouse server. "); + clientPing.close(); + return false; + } + + public ClickHouseNode getServer() { + return this.server; + } + + public List<ClickHouseRecord> query(String query) { + return query(query, ClickHouseFormat.RowBinaryWithNamesAndTypes); + } + + public List<ClickHouseRecord> query(String query, ClickHouseFormat clickHouseFormat) { + int retryCount = 0; + Exception ce = null; + while (retryCount < retry) { + try (ClickHouseClient client = ClickHouseClient.newInstance(ClickHouseProtocol.HTTP); + ClickHouseResponse response = client.read(server) + .format(clickHouseFormat) + .query(query) + .execute().get()) { + + List<ClickHouseRecord> recordList = new ArrayList<>(); + for (ClickHouseRecord r : response.records()) { + recordList.add(r); + } + return recordList; + + } catch (Exception e) { + retryCount++; + LOGGER.warn(String.format("Query retry %d out of %d", retryCount, retry), e); + ce = e; + } + } + throw new RuntimeException(ce); + + } + + private Connection getConnection(String url, Properties properties) throws SQLException { + ClickHouseDataSource dataSource = new ClickHouseDataSource(url, properties); + Connection conn = dataSource.getConnection(config.getUserName(), config.getPassWord()); + + System.out.println("Connected to: " + conn.getMetaData().getURL()); + return conn; + } + + private boolean insertJson(String jsonString, String table, String sql, String url) { + + try (Connection connection = getConnection(url, new Properties()); + PreparedStatement ps = connection.prepareStatement(sql)) { + ps.setObject(1, new ClickHouseWriter() { + @Override + public void write(ClickHouseOutputStream output) throws IOException { + output.writeBytes(jsonString.getBytes()); + } + }); + ps.executeUpdate(); + Review Comment: Infinite loop bug in insertJson: retryCount is never incremented. The loop `while (retryCount < this.retry)` will spin forever if insertJson returns false, causing the thread to hang indefinitely. ########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkConnector.java: ########## @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.sink; + +import io.openmessaging.KeyValue; +import io.openmessaging.connector.api.component.task.Task; +import io.openmessaging.connector.api.component.task.sink.SinkConnector; +import java.util.ArrayList; +import java.util.List; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSinkConfig; + +public class ClickHouseSinkConnector extends SinkConnector { + + private KeyValue keyValue; + + @Override public List<KeyValue> taskConfigs(int maxTasks) { + List<KeyValue> configs = new ArrayList<>(); + for (int i = 0; i < maxTasks; i++) { + configs.add(this.keyValue); + } Review Comment: taskConfigs adds the same shared KeyValue reference to every task slot. If tasks are run concurrently and any task mutates the KeyValue object, all tasks will be affected. Each task should receive an independent copy of the configuration. ########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java: ########## @@ -0,0 +1,168 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.helper; + +import com.clickhouse.client.ClickHouseClient; +import com.clickhouse.client.ClickHouseCredentials; +import com.clickhouse.client.ClickHouseNode; +import com.clickhouse.client.ClickHouseProtocol; +import com.clickhouse.client.ClickHouseResponse; +import com.clickhouse.data.ClickHouseFormat; +import com.clickhouse.data.ClickHouseOutputStream; +import com.clickhouse.data.ClickHouseRecord; +import com.clickhouse.data.ClickHouseWriter; +import com.clickhouse.jdbc.ClickHouseDataSource; +import java.io.IOException; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ClickHouseHelperClient { + + private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class); + + private ClickHouseBaseConfig config; + private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC; + private ClickHouseNode server = null; + private int retry = ClickHouseConstants.retryCountDefault; + + public ClickHouseHelperClient(ClickHouseBaseConfig config) { + this.config = config; + this.server = create(config); + } + + private ClickHouseNode create(ClickHouseBaseConfig config) { + this.server = ClickHouseNode.builder() + .host(config.getClickHouseHost()) + .port(ClickHouseProtocol.HTTP, config.getClickHousePort()) + .database(config.getDatabase()).credentials(getCredentials(config)) + .build(); + + return this.server; + } + + private ClickHouseCredentials getCredentials(ClickHouseBaseConfig config) { + if (config.getUserName() != null && config.getPassWord() != null) { + return ClickHouseCredentials.fromUserAndPassword(config.getUserName(), config.getPassWord()); + } + if (config.getAccessToken() != null) { + return ClickHouseCredentials.fromAccessToken(config.getAccessToken()); + } + throw new RuntimeException("Credentials cannot be empty!"); Review Comment: ping() creates a new ClickHouseClient on every call but only closes it on success. If ping returns false after all retries, the client opened in the last failed iteration is never closed, leaking a connection resource. ########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseBaseConfig.java: ########## @@ -0,0 +1,138 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.config; + +import io.openmessaging.KeyValue; +import java.lang.reflect.Method; + +public class ClickHouseBaseConfig { + + private String clickHouseHost; + + private Integer clickHousePort; + + private String database; + + private String userName; + + private String passWord; + + private String accessToken; + + private String topic; + + public String getTopic() { + return topic; + } + + public void setTopic(String topic) { + this.topic = topic; + } + + public String getClickHouseHost() { + return clickHouseHost; + } + + public void setClickHouseHost(String clickHouseHost) { + this.clickHouseHost = clickHouseHost; + } + + public Integer getClickHousePort() { + return clickHousePort; + } + + public void setClickHousePort(Integer clickHousePort) { + this.clickHousePort = clickHousePort; + } + + public String getUserName() { + return userName; + } + + public void setUserName(String userName) { + this.userName = userName; + } + + public String getPassWord() { + return passWord; + } + + public void setPassWord(String passWord) { + this.passWord = passWord; + } + + public String getAccessToken() { + return accessToken; + } + + public void setAccessToken(String accessToken) { + this.accessToken = accessToken; + } + + public String getDatabase() { + return database; + } + + public void setDatabase(String database) { + this.database = database; + } + + public void load(KeyValue props) { + properties2Object(props, this); + } + + private void properties2Object(final KeyValue p, final Object object) { + + Method[] methods = object.getClass().getMethods(); + for (Method method : methods) { + String mn = method.getName(); + if (mn.startsWith("set")) { + try { Review Comment: properties2Object silently ignores all exceptions via `catch (Throwable ignored)`. If a required config value fails to parse (e.g. a non-numeric string for clickhouseport), the field remains null/default and no error is surfaced, leading to a confusing NPE or connection failure later. ########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/sink/ClickHouseSinkTask.java: ########## @@ -0,0 +1,80 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.sink; + +import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; +import io.openmessaging.KeyValue; +import io.openmessaging.connector.api.component.task.sink.SinkTask; +import io.openmessaging.connector.api.data.ConnectRecord; +import io.openmessaging.connector.api.data.Field; +import io.openmessaging.connector.api.data.Struct; +import io.openmessaging.connector.api.errors.ConnectException; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import org.apache.rocketmq.connect.clickhouse.helper.ClickHouseHelperClient; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSinkConfig; + +public class ClickHouseSinkTask extends SinkTask { + + public ClickHouseSinkConfig config; + + private ClickHouseHelperClient helperClient; + + @Override public void put(List<ConnectRecord> sinkRecords) throws ConnectException { + if (sinkRecords == null || sinkRecords.size() < 1) { + return; + } + Map<String, JSONArray> valueMap = new HashMap<>(); + for (ConnectRecord record : sinkRecords) { + String table = record.getSchema().getName(); + JSONArray jsonArray = valueMap.getOrDefault(table, new JSONArray()); + + final List<Field> fields = record.getSchema().getFields(); + final Struct structData = (Struct) record.getData(); + + JSONObject object = new JSONObject(); + for (Field field : fields) { + object.put(field.getName(), structData.get(field)); Review Comment: record.getData() is cast unconditionally to Struct without any null or type check. If the connector receives a non-Struct payload (e.g. a plain String or Map from a JSON converter without schema), this will throw a ClassCastException and fail the entire batch silently because ConnectException is caught by the framework but the root cause may be swallowed. ########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/helper/ClickHouseHelperClient.java: ########## @@ -0,0 +1,168 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.helper; + +import com.clickhouse.client.ClickHouseClient; +import com.clickhouse.client.ClickHouseCredentials; +import com.clickhouse.client.ClickHouseNode; +import com.clickhouse.client.ClickHouseProtocol; +import com.clickhouse.client.ClickHouseResponse; +import com.clickhouse.data.ClickHouseFormat; +import com.clickhouse.data.ClickHouseOutputStream; +import com.clickhouse.data.ClickHouseRecord; +import com.clickhouse.data.ClickHouseWriter; +import com.clickhouse.jdbc.ClickHouseDataSource; +import java.io.IOException; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.SQLException; +import java.util.ArrayList; +import java.util.List; +import java.util.Properties; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseConstants; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseBaseConfig; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +public class ClickHouseHelperClient { + + private static final Logger LOGGER = LoggerFactory.getLogger(ClickHouseHelperClient.class); + + private ClickHouseBaseConfig config; + private int timeout = ClickHouseConstants.timeoutSecondsDefault * ClickHouseConstants.MILLI_IN_A_SEC; + private ClickHouseNode server = null; + private int retry = ClickHouseConstants.retryCountDefault; + + public ClickHouseHelperClient(ClickHouseBaseConfig config) { + this.config = config; + this.server = create(config); + } + + private ClickHouseNode create(ClickHouseBaseConfig config) { + this.server = ClickHouseNode.builder() + .host(config.getClickHouseHost()) + .port(ClickHouseProtocol.HTTP, config.getClickHousePort()) + .database(config.getDatabase()).credentials(getCredentials(config)) + .build(); + + return this.server; + } Review Comment: getCredentials throws a RuntimeException when neither username/password nor accessToken is set, but this is thrown inside the constructor during create(). The error message is not descriptive enough to guide operators. More importantly, a missing-credential config should be caught at connector validation time (in start()), not deferred to client construction. ########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseSinkConfig.java: ########## @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.config; + +import java.util.HashSet; +import java.util.Set; + +public class ClickHouseSinkConfig extends ClickHouseBaseConfig { + public static final Set<String> SINK_REQUEST_CONFIG = new HashSet<String>() { + { Review Comment: SINK_REQUEST_CONFIG only requires clickhousehost and clickhouseport, omitting database, username, and password. A connector started without credentials will pass validation but fail at runtime in getCredentials(). Minimum required fields should include database, username, and password (or accesstoken). ########## connectors/rocketmq-connect-clickhouse/src/main/java/org/apache/rocketmq/connect/clickhouse/source/ClickHouseSourceConnector.java: ########## @@ -0,0 +1,57 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.rocketmq.connect.clickhouse.source; + +import io.openmessaging.KeyValue; +import io.openmessaging.connector.api.component.task.Task; +import io.openmessaging.connector.api.component.task.source.SourceConnector; +import java.util.ArrayList; +import java.util.List; +import org.apache.rocketmq.connect.clickhouse.config.ClickHouseSourceConfig; + +public class ClickHouseSourceConnector extends SourceConnector { + + private KeyValue keyValue; + + @Override public List<KeyValue> taskConfigs(int maxTasks) { + List<KeyValue> configs = new ArrayList<>(); + for (int i = 0; i < maxTasks; i++) { + configs.add(this.keyValue); + } Review Comment: Same shared-reference issue as ClickHouseSinkConnector.taskConfigs: all task slots receive the same KeyValue object reference. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
