[
https://issues.apache.org/jira/browse/CAMEL-24245?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18098314#comment-18098314
]
Omar Atie commented on CAMEL-24245:
-----------------------------------
**Implementation summary**
Added the new `camel-clickhouse` component (producer-only) using the official
ClickHouse Java client (`com.clickhouse:client-v2`). Supported operations:
- `insert` (default) — stream data in native formats (`JSONEachRow`,
`RowBinary`, `CSV`, etc.)
- `query` — run SQL and return results in the configured format
- `ping` — health check
**PR:** https://github.com/apache/camel/pull/25034
---
**Live validation (ClickHouse 24.8 + Camel routes)**
Validated all use cases against a live ClickHouse server (Docker:
`mirror.gcr.io/clickhouse/clickhouse-server:24.8`) using standalone Camel
routes. No Apache Camel repo changes were made for this validation; a separate
demo project was used (`C:\c\camel-clickhouse-demo`).
**Environment**
- ClickHouse: `http://localhost:8123`
- Database/table: `camel_demo.events`
- Schema: `(id UInt32, name String, source String DEFAULT 'test') ENGINE =
MergeTree ORDER BY id`
- Auth: `default` / empty password
**Use cases tested — all 11 passed**
| # | Use case | Route / config | DB validation |
|---|----------|----------------|---------------|
| 1 | ping | `operation=ping` | `CamelClickHousePingOk=true` |
| 2 | insert String | `format=JSONEachRow`, String body | `id=1, source=string`
|
| 3 | insert byte[] | JSONEachRow bytes | `id=2, source=bytes` |
| 4 | insert File | JSONEachRow from file | `id=3, source=carol/file` |
| 5 | insert CSV | `format=CSV` | `id=4, source=csv` |
| 6 | query count | SQL in body, `format=CSV` | count=4 at step 6 |
| 7 | query select | `SELECT id, name WHERE id=1` | returns `1,"alice"` |
| 8 | asyncInsert | `asyncInsert=true&waitForAsyncInsert=true` | `id=5,
source=async` present in DB |
| 9 | header overrides | `CamelClickHouseTable` + `CamelClickHouseFormat`
headers | `id=6, source=headers` |
| 10 | batchSize + List | `batchSize=2`, 5 registered POJOs → 3 inserts | 5
rows `id=10..14, source=batch` |
| 11 | default operation | no `operation` param (defaults to insert) | `id=7,
source=default-op` |
**Final DB state (validated via `clickhouse-client`)**
```
12 rows in camel_demo.events
By source:
string, bytes, file, csv, async, headers, default-op → 1 row each
batch → 5 rows (batchSize=2 split 5 POJOs into 3 insert calls)
```
---
**Observations / notes for reviewers**
1. **`batchSize`** applies to **List** bodies only. Requires a shared `Client`
with the POJO class registered via `client.register(Class,
client.getTableSchema(table))`. Stream/String/byte[]/File inserts are
unaffected.
2. **`asyncInsert=true`** — data is written correctly, but
`CamelClickHouseWrittenRows` may be **0** until the server flushes async
buffers. Validate with a query, not only the header.
3. **ClickHouse 24.8 Docker** — recent images require explicit auth config
(`CLICKHOUSE_USER=default`, `CLICKHOUSE_PASSWORD=`,
`CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT=1`) for programmatic clients.
4. **Review feedback addressed in PR** — `volatile` lazy client, `ssl=true` →
`https://`, `skipITs.ppc64le`/`skipITs.s390x`, unrelated upgrade-guide section
reverted, client-side batching implemented.
---
**Tests**
- Unit tests: 16 passed (AssertJ + Mockito)
- Integration tests: 2 (Testcontainers, skipped without Docker in local run)
- Live demo: 11/11 use cases passed with DB validation
---
**Example routes**
```java
// Insert JSONEachRow
from("direct:insert")
.to("clickhouse://camel_demo.events?operation=insert&format=JSONEachRow&serverUrl=http://localhost:8123";);
// Query
from("direct:query")
.setBody(constant("SELECT count() FROM camel_demo.events"))
.to("clickhouse://camel_demo?operation=query&format=CSV&serverUrl=http://localhost:8123";);
// Ping
from("direct:ping")
.to("clickhouse://camel_demo?operation=ping&serverUrl=http://localhost:8123";);
// Batch List insert (shared Client with registered POJO required)
from("direct:insertBatch")
.to("clickhouse://camel_demo.events?operation=insert&batchSize=2");
```
---
_AI-generated comment prepared on behalf of the contributor._
---
Want a shorter version, or should I tailor it (e.g. drop the demo section and
keep only PR/test summary)?
> Camel-ClickHouse New Component Proposal
> ---------------------------------------
>
> Key: CAMEL-24245
> URL: https://issues.apache.org/jira/browse/CAMEL-24245
> Project: Camel
> Issue Type: New Feature
> Reporter: Omar Atie
> Assignee: Omar Atie
> Priority: Major
>
> I'd like to propose a new component for integrating with *ClickHouse*, the
> high-performance columnar OLAP database.
> Camel can talk to ClickHouse today through the generic xref camel-jdbc /
> camel-sql components, but only over the JDBC PreparedStatement path. That
> works for low-volume CRUD, but it leaves ClickHouse's high-throughput
> ingestion features on the table: native RowBinary/format streaming inserts,
> server-side asynchronous inserts, bulk load from files, and compression.
> Users building analytics and observability pipelines currently hand-roll
> beans around the ClickHouse client to get acceptable ingest performance.
> The idea is a camel-clickhouse component built on the official ClickHouse
> Java client (client-v2, com.clickhouse, available in Maven Central — the same
> library that backs the ClickHouse JDBC driver) that would expose ClickHouse's
> native capabilities as first-class endpoint options.
> clickhouse://my_db.events?operation=insert&format=RowBinary&batchSize=5000
> This follows the pattern already used by camel-influxdb2 (a dedicated
> component on a vendor client, rather than generic JDBC), which is the closest
> analogue in the catalog.
> h2. *Why a dedicated component (vs camel-jdbc)*
> - *Native batch insert* via \{{Client.insert(table, List<?>,
> InsertSettings)}} and RowBinary — significantly faster than JDBC
> \{{addBatch()/executeBatch()}} for large volumes.
> - *Asynchronous inserts* (\{{async_insert=1}}) for high-concurrency,
> small-batch ingestion without client-side buffering.
> - *Format streaming* — stream JSONEachRow / CSV / TSV / Parquet bodies
> straight to the server with no per-row serialization.
> - *Bulk load from files* (\{{INSERT ... FROM INFILE}}) with compression
> (lz4/zstd).
> - *Idiomatic options* — database, table, format, batchSize, compression,
> async — instead of opaque JDBC URL params.
> h2. *Design*
> - *Producer-only* (like camel-jdbc): ClickHouse is ingest-via-producer;
> OLAP querying is request/reply.
> - *Operations:* \{{insert}} (default), \{{query}}, \{{ping}}.
> - *Body types accepted for insert:* \{{List<Map<String,Object>>}},
> \{{List<POJO>}}, JSON/CSV/TSV String or InputStream (matched to \{{format}}),
> or a \{{java.io.File}} for bulk load.
> - *Client sharing:* autowire a shared \{{com.clickhouse.client.api.Client}}
> bean, or configure \{{serverUrl}}/\{{username}}/\{{password}} on the endpoint.
> - *Tests:* ClickHouse Testcontainers via a new
> \{{camel-test-infra-clickhouse}} module; AssertJ assertions.
> h2. *Use Cases*
> {*}Use Case 1: High-throughput event ingestion from Kafka\{*}
> Stream events from Kafka and batch-insert them into ClickHouse using the
> native RowBinary format for maximum ingest performance.
> {code:java}
> from("kafka:events?groupId=analytics")
> .aggregate(constant(true), new GroupedBodyAggregationStrategy())
> .completionSize(5000).completionTimeout(2000)
>
> .to("clickhouse://analytics.events?operation=insert&format=RowBinary&batchSize=5000")
> .log("Inserted ${header.CamelClickHouseWrittenRows} rows");
> {code}
> {*}Use Case 2: Server-side asynchronous inserts for many small producers\{*}
> Let ClickHouse buffer and flush inserts server-side, ideal for many
> concurrent producers sending small payloads.
> {code:java}
> from("platform-http:/ingest")
> .unmarshal().json()
>
> .to("clickhouse://metrics.samples?operation=insert&asyncInsert=true&waitForAsyncInsert=false");
> {code}
> {*}Use Case 3: Scheduled OLAP query feeding a dashboard/alert\{*}
> Run an aggregation query on a timer and route the result set to downstream
> systems.
> {code:java}
> from("timer:rollup?period=60000")
> .setBody(constant(
> "SELECT toStartOfMinute(ts) AS minute, count() AS hits " +
> "FROM analytics.events WHERE ts > now() - INTERVAL 5 MINUTE " +
> "GROUP BY minute ORDER BY minute"))
> .to("clickhouse://analytics?operation=query&format=JSONEachRow")
> .to("kafka:rollup-metrics");
> {code}
> {*}Use Case 4: Bulk load from files (CSV/Parquet) with compression\{*}
> Ingest data files dropped into a directory using ClickHouse's native file
> load with zstd compression.
> {code:java}
> from("file:data/incoming?include=.*\\.csv.zst&move=.done")
>
> .to("clickhouse://warehouse.orders?operation=insert&format=CSV&compression=zstd")
> .log("Loaded file ${header.CamelFileName} into ClickHouse");
> {code}
> {*}Use Case 5: ETL — migrate/aggregate from OLTP into ClickHouse\{*}
> Read rows from a relational source and continuously roll them into ClickHouse
> for analytics, decoupling reporting load from the OLTP database.
> {code:java}
> from("sql:SELECT * FROM orders WHERE exported = false?dataSource=#pg")
> .split(body()).streaming()
> .aggregate(constant(true), new GroupedBodyAggregationStrategy())
> .completionSize(10000).completionTimeout(5000)
>
> .to("clickhouse://warehouse.orders_fact?operation=insert&format=JSONEachRow");
> {code}
> {*}Use Case 6: Observability — write application/access logs to ClickHouse\{*}
> Fan structured log events into ClickHouse as a cost-effective, queryable log
> store.
> {code:java}
> from("direct:appLog")
> .marshal().json()
>
> .to("clickhouse://logs.app_logs?operation=insert&format=JSONEachRow&asyncInsert=true");
> {code}
> {*}Use Case 7: Health check / readiness probe\{*}
> Verify connectivity to the ClickHouse cluster before a route starts
> processing.
> {code:java}
> from("timer:health?period=30000")
> .to("clickhouse://default?operation=ping")
> .choice()
> .when(header("CamelClickHousePingOk").isEqualTo(true))
> .to("direct:markHealthy")
> .otherwise()
> .to("direct:alertOps")
> .end();
> {code}
> h2. *Proposed URI options (initial)*
> - \{{serverUrl}} — ClickHouse HTTP endpoint (e.g. http://localhost:8123),
> or autowire a shared Client bean
> - \{{database}} / table via path — \{{clickhouse://<database>.<table>}}
> - \{{operation}} — insert | query | ping (default: insert)
> - \{{format}} — RowBinary | JSONEachRow | CSV | TSV | Parquet ... (default:
> JSONEachRow)
> - \{{batchSize}} — client-side batch size for insert
> - \{{asyncInsert}} / \{{waitForAsyncInsert}} — server-side async insert
> - \{{compression}} — none | lz4 | zstd
> - \{{username}} / \{{password}} (secret) / \{{ssl}}
> h2. *Proposed message headers*
> - \{{CamelClickHouseOperation}} — override the endpoint operation
> - \{{CamelClickHouseDatabase}} / \{{CamelClickHouseTable}} — override target
> - \{{CamelClickHouseFormat}} — override format
> - \{{CamelClickHouseWrittenRows}} — (out) rows written on insert
> - \{{CamelClickHousePingOk}} — (out) boolean result of a ping
> I'm happy to implement this and follow the camel-influxdb2 layout, add a
> camel-test-infra-clickhouse module with Testcontainers, and provide docs + an
> upgrade-guide entry. Feedback on the operation set and default format is
> welcome.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
