Worker crashes with TOO_MANY_SIMULTANEOUS_QUERIES exception

[danifullview]

Describe the bug

Kafka ClickHouse sink workers crash under high traffic.

Steps to reproduce

1. Create 6 workers in 6 pods on k8s
2. Push high volumes of data

Expected behaviour

Error log

org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:618)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
	at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.RuntimeException: java.util.concurrent.ExecutionException: com.clickhouse.client.ClickHouseException: Code: 202. DB::Exception: Too many simultaneous queries. Maximum: 1000. (TOO_MANY_SIMULTANEOUS_QUERIES) (version 23.3.1.30147 (official build))
, server ClickHouseNode [uri=https://r0d3431wy4.eu-central-1.aws.clickhouse.cloud:8443/fullview, options={sslmode=STRICT}]
	at com.clickhouse.kafka.connect.util.Utils.handleException(Utils.java:89)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsert(ClickHouseWriter.java:143)
	at com.clickhouse.kafka.connect.sink.processing.Processing.doInsert(Processing.java:45)
	at com.clickhouse.kafka.connect.sink.processing.Processing.doLogic(Processing.java:151)
	at com.clickhouse.kafka.connect.sink.ProxySinkTask.put(ProxySinkTask.java:80)
	at com.clickhouse.kafka.connect.sink.ClickHouseSinkTask.put(ClickHouseSinkTask.java:60)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:587)
	... 11 more
Caused by: java.util.concurrent.ExecutionException: com.clickhouse.client.ClickHouseException: Code: 202. DB::Exception: Too many simultaneous queries. Maximum: 1000. (TOO_MANY_SIMULTANEOUS_QUERIES) (version 23.3.1.30147 (official build))
, server ClickHouseNode [uri=https://r0d3431wy4.eu-central-1.aws.clickhouse.cloud:8443/fullview, options={sslmode=STRICT}]
	at java.base/java.util.concurrent.CompletableFuture.reportGet(CompletableFuture.java:395)
	at java.base/java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1999)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsertJson(ClickHouseWriter.java:472)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsert(ClickHouseWriter.java:138)
	... 16 more
Caused by: com.clickhouse.client.ClickHouseException: Code: 202. DB::Exception: Too many simultaneous queries. Maximum: 1000. (TOO_MANY_SIMULTANEOUS_QUERIES) (version 23.3.1.30147 (official build))
, server ClickHouseNode [uri=https://r0d3431wy4.eu-central-1.aws.clickhouse.cloud:8443/fullview, options={sslmode=STRICT}]
	at com.clickhouse.client.ClickHouseException.of(ClickHouseException.java:148)
	at com.clickhouse.client.AbstractClient.lambda$execute$0(AbstractClient.java:291)
	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700)
	... 3 more
Caused by: java.io.IOException: Code: 202. DB::Exception: Too many simultaneous queries. Maximum: 1000. (TOO_MANY_SIMULTANEOUS_QUERIES) (version 23.3.1.30147 (official build))

	at com.clickhouse.client.http.HttpUrlConnectionImpl.checkResponse(HttpUrlConnectionImpl.java:187)
	at com.clickhouse.client.http.HttpUrlConnectionImpl.post(HttpUrlConnectionImpl.java:278)
	at com.clickhouse.client.http.ClickHouseHttpClient.send(ClickHouseHttpClient.java:101)
	at com.clickhouse.client.AbstractClient.sendAsync(AbstractClient.java:156)
	at com.clickhouse.client.AbstractClient.lambda$execute$0(AbstractClient.java:289)
	... 4 more

Configuration

Environment

• Kafka-Connect version: 7.4.0
• Kafka Connect configuration:
• Kafka version: Confluent cloud
• Kafka environment: -
• OS: -

ClickHouse server

• ClickHouse Server version: 23.3.1.30147 (cloud)
• ClickHouse Server non-default settings, if any:

[mzitnik]

@danifullview did #110 helped?

[danifullview]

@mzitnik thanks for #110. I got it and deployed it, but got sidetracked by another issue and did not get the change to test it yet.
Will test it later today and let you know.

Thank you.

[danifullview]

I'm getting another exception now:

org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:618)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
	at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.RuntimeException: com.clickhouse.client.ClickHouseException: Read timed out, server ClickHouseNode [uri=https://r0d3431wy4.eu-central-1.aws.clickhouse.cloud:8443/fullview, options={sslmode=STRICT}]
	at com.clickhouse.kafka.connect.sink.state.provider.KeeperStateProvider.getStateRecord(KeeperStateProvider.java:96)
	at com.clickhouse.kafka.connect.sink.processing.Processing.doLogic(Processing.java:97)
	at com.clickhouse.kafka.connect.sink.ProxySinkTask.put(ProxySinkTask.java:80)
	at com.clickhouse.kafka.connect.sink.ClickHouseSinkTask.put(ClickHouseSinkTask.java:60)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:587)
	... 11 more
Caused by: com.clickhouse.client.ClickHouseException: Read timed out, server ClickHouseNode [uri=https://r0d3431wy4.eu-central-1.aws.clickhouse.cloud:8443/fullview, options={sslmode=STRICT}]
	at com.clickhouse.client.ClickHouseException.of(ClickHouseException.java:144)
	at com.clickhouse.client.AbstractClient.lambda$execute$0(AbstractClient.java:291)
	at java.base/java.util.concurrent.CompletableFuture$AsyncSupply.run(CompletableFuture.java:1700)
	... 3 more
Caused by: java.net.SocketTimeoutException: Read timed out
	at java.base/java.net.SocketInputStream.socketRead0(Native Method)
	at java.base/java.net.SocketInputStream.socketRead(SocketInputStream.java:115)
	at java.base/java.net.SocketInputStream.read(SocketInputStream.java:168)
	at java.base/java.net.SocketInputStream.read(SocketInputStream.java:140)
	at java.base/sun.security.ssl.SSLSocketInputRecord.read(SSLSocketInputRecord.java:484)
	at java.base/sun.security.ssl.SSLSocketInputRecord.readHeader(SSLSocketInputRecord.java:478)
	at java.base/sun.security.ssl.SSLSocketInputRecord.bytesInCompletePacket(SSLSocketInputRecord.java:70)
	at java.base/sun.security.ssl.SSLSocketImpl.readApplicationRecord(SSLSocketImpl.java:1455)
	at java.base/sun.security.ssl.SSLSocketImpl$AppInputStream.read(SSLSocketImpl.java:1066)
	at java.base/java.io.BufferedInputStream.fill(BufferedInputStream.java:252)
	at java.base/java.io.BufferedInputStream.read1(BufferedInputStream.java:292)
	at java.base/java.io.BufferedInputStream.read(BufferedInputStream.java:351)
	at java.base/sun.net.www.http.HttpClient.parseHTTPHeader(HttpClient.java:787)
	at java.base/sun.net.www.http.HttpClient.parseHTTP(HttpClient.java:722)
	at java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream0(HttpURLConnection.java:1620)
	at java.base/sun.net.www.protocol.http.HttpURLConnection.getInputStream(HttpURLConnection.java:1525)
	at java.base/java.net.HttpURLConnection.getResponseCode(HttpURLConnection.java:527)
	at java.base/sun.net.www.protocol.https.HttpsURLConnectionImpl.getResponseCode(HttpsURLConnectionImpl.java:334)
	at com.clickhouse.client.http.HttpUrlConnectionImpl.checkResponse(HttpUrlConnectionImpl.java:156)
	at com.clickhouse.client.http.HttpUrlConnectionImpl.post(HttpUrlConnectionImpl.java:278)
	at com.clickhouse.client.http.ClickHouseHttpClient.send(ClickHouseHttpClient.java:101)
	at com.clickhouse.client.AbstractClient.sendAsync(AbstractClient.java:156)
	at com.clickhouse.client.AbstractClient.lambda$execute$0(AbstractClient.java:289)
	... 4 more

[danifullview]

Another error that killed the task:

org.apache.kafka.connect.errors.ConnectException: Exiting WorkerSinkTask due to unrecoverable exception.
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:618)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.poll(WorkerSinkTask.java:336)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.iteration(WorkerSinkTask.java:237)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.execute(WorkerSinkTask.java:206)
	at org.apache.kafka.connect.runtime.WorkerTask.doRun(WorkerTask.java:202)
	at org.apache.kafka.connect.runtime.WorkerTask.run(WorkerTask.java:257)
	at org.apache.kafka.connect.runtime.isolation.Plugins.lambda$withClassLoader$1(Plugins.java:181)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
	at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
	at java.base/java.lang.Thread.run(Thread.java:829)
Caused by: java.lang.RuntimeException: java.io.IOException: Write timed out after 30000 ms
	at com.clickhouse.kafka.connect.util.Utils.handleException(Utils.java:91)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsert(ClickHouseWriter.java:143)
	at com.clickhouse.kafka.connect.sink.processing.Processing.doInsert(Processing.java:45)
	at com.clickhouse.kafka.connect.sink.processing.Processing.doLogic(Processing.java:151)
	at com.clickhouse.kafka.connect.sink.ProxySinkTask.put(ProxySinkTask.java:80)
	at com.clickhouse.kafka.connect.sink.ClickHouseSinkTask.put(ClickHouseSinkTask.java:60)
	at org.apache.kafka.connect.runtime.WorkerSinkTask.deliverMessages(WorkerSinkTask.java:587)
	... 11 more
Caused by: java.io.IOException: Write timed out after 30000 ms
	at com.clickhouse.client.stream.BlockingPipedOutputStream.updateBuffer(BlockingPipedOutputStream.java:66)
	at com.clickhouse.client.stream.BlockingPipedOutputStream.updateBuffer(BlockingPipedOutputStream.java:55)
	at com.clickhouse.client.stream.BlockingPipedOutputStream.writeBytes(BlockingPipedOutputStream.java:180)
	at com.clickhouse.client.ClickHouseOutputStream.write(ClickHouseOutputStream.java:231)
	at com.clickhouse.client.data.BinaryStreamUtils.writeBytes(BinaryStreamUtils.java:234)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsertJson(ClickHouseWriter.java:463)
	at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsert(ClickHouseWriter.java:138)
	... 16 more
	Suppressed: java.io.IOException: Close stream timed out after 30000 ms
		at com.clickhouse.client.stream.BlockingPipedOutputStream.close(BlockingPipedOutputStream.java:96)
		at com.clickhouse.kafka.connect.sink.db.ClickHouseWriter.doInsertJson(ClickHouseWriter.java:447)
		... 17 more

[danifullview]

After I restart the workers, most of the time I get the error described in #103 .

[Paultagoras]

We're expanding the list of error codes that the connector retries - that should help to add some redundancy.

[mzitnik]

Implemented a retry mechanism for that specific error can we found in latest release.