Skip to content

[ISSUE #478] Support ClickHouse source and sink #496

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
joeCarf wants to merge 9 commits into
base: master
Choose a base branch
from
Open

[ISSUE #478] Support ClickHouse source and sink #496

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
9 commits
Select commit Hold shift + click to select a range
5117ede
initial commit
joeCarf May 5, 2023
25e7ba5
commit
joeCarf May 5, 2023
54c2517
rename
joeCarf May 5, 2023
1957ce8
finish source connector
joeCarf May 6, 2023
eb5033d
finish sink connector
joeCarf May 6, 2023
969b395
add license
joeCarf May 6, 2023
ca0d454
Update README.md
joeCarf May 6, 2023
31cd5e4
Update connect-standalone.conf
joeCarf May 6, 2023
d5c7461
Merge branch 'apache:master' into dev/clickhouse-source
joeCarf Jul 25, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
51 changes: 51 additions & 0 deletions connectors/rocketmq-connect-clickhouse/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
##### ClickHouseSourceConnector fully-qualified name

@RockteMQ-AI RockteMQ-AI Aug 5, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

org.apache.rocketmq.connect.clickhouse.source.ClickHouseSourceConnector

**clickhouse-source-connector** start

```
POST http://${runtime-ip}:${runtime-port}/connectors/clickhouseSourceConnector
{
"connector.class":"org.apache.rocketmq.connect.clickhouse.source.ClickHouseSourceConnector",
"clickhousehost":"localhost",
"clickhouseport":8123,
"database":"default",
"username":"default",
"password":"123456",
"table":"tableName",
"topic":"testClickHouseTopic",
"value.converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
"key.converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
}
```

##### ClickHouseSinkConnector fully-qualified name
org.apache.rocketmq.connect.clickhouse.sink.ClickHouseSinkConnector

**clickhouse-sink-connector** start

```
POST http://${runtime-ip}:${runtime-port}/connectors/clickhouseSinkConnector
{
"connector.class":"org.apache.rocketmq.connect.clickhouse.sink.ClickHouseSinkConnector",
"clickhousehost":"localhost",
"clickhouseport":8123,
"database":"clickhouse",
"username":"default",
"password":"123456",
"connect.topicnames":"testClickHouseTopic",
"value.converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter",
"key.converter":"org.apache.rocketmq.connect.runtime.converter.record.json.JsonConverter"
}
```

##### parameter configuration

| parameter | effect | required | default |
|--------------------|---------------------------------------------------|-------------------|---------|
| clickhousehost | The Host of the Clickhouse server | yes | null |
| clickhouseport | The Port of the Clickhouse server | yes | null |
| database | The database to read or write | yes | null |
| table | The source table to read | yes (source only) | null |
| topic | RocketMQ topic for source connector to write into | yes (source only) | null |
| connect.topicnames | RocketMQ topic for sink connector to read from | yes (sink only) | null |
204 changes: 204 additions & 0 deletions connectors/rocketmq-connect-clickhouse/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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>

@RockteMQ-AI RockteMQ-AI Aug 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

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

@RockteMQ-AI RockteMQ-AI Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

<!-- use uber jar with all dependencies included, change classifier to http for smaller jar -->
<classifier>all</classifier>
</dependency>

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

@RockteMQ-AI RockteMQ-AI Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

</dependency>

<dependency>
<groupId>com.clickhouse</groupId>
<!-- or clickhouse-grpc-client if you prefer gRPC -->
<artifactId>clickhouse-http-client</artifactId>
<version>0.4.5</version>
</dependency>
<dependency>
<groupId>com.alibaba</groupId>
<artifactId>fastjson</artifactId>
<version>1.2.83</version>
<scope>compile</scope>
</dependency>
<dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>RELEASE</version>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>1.7.7</version>
</dependency>
</dependencies>

</project>
138 changes: 138 additions & 0 deletions ...use/src/main/java/org/apache/rocketmq/connect/clickhouse/config/ClickHouseBaseConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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;

@RockteMQ-AI RockteMQ-AI Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The database field has no default value and is not in the required-config sets (REQUEST_CONFIG / SINK_REQUEST_CONFIG). If omitted, ClickHouseNode.builder().database(null) and the JDBC URL will contain a null database, causing connection failures. Either make it required or provide a sensible default (e.g., "default").

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

@RockteMQ-AI RockteMQ-AI Aug 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

}

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

@RockteMQ-AI RockteMQ-AI Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Method[] methods = object.getClass().getMethods();
for (Method method : methods) {
String mn = method.getName();
if (mn.startsWith("set")) {
try {

@RockteMQ-AI RockteMQ-AI Aug 23, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

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);
}

@RockteMQ-AI RockteMQ-AI Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

}
} catch (Throwable ignored) {
}
}
}
}
}
Loading