nzw921rx opened a new issue, #10953: URL: https://github.com/apache/seatunnel/issues/10953
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues. ### What happened When using the Kafka source connector in streaming mode, the SeaTunnel worker process experienced extremely high heap memory usage (1.2 GB out of 1.3 GB total), causing frequent Full GC and OOM. After capturing a heap dump and analyzing it with Eclipse MAT, the root cause was identified: A single `java.util.concurrent.LinkedBlockingQueue` instance occupied **1,284,433,784 bytes (94.31%)** of the heap. This queue is the internal element queue between the Kafka fetcher thread and the reader thread, created in `KafkaSource#createReader()`. #### Root Cause The default value of `reader_cache_queue_size` is **1024**, and each element in the queue is a **complete `consumer.poll()` batch** (not a single message). With Kafka's default `max.poll.records = 500`, the worst-case buffer size is: ``` 1024 batches × 500 messages/batch × message_size ``` In our case, each message is approximately 12 KB, so: ``` 1024 × 500 × 12 KB ≈ 6 GB (theoretical max) ``` The queue was partially filled, already consuming 1.2 GB. When downstream sink processing is slower than the Kafka fetch rate (back-pressure), the queue fills up rapidly. #### Heap Dump Analysis <img width="1827" height="1298" alt="Image" src="https://github.com/user-attachments/assets/c2fa54b0-ff98-4a3f-ad20-bc7255e90c7f"; /> ### SeaTunnel Version 2.3.13 ### SeaTunnel Config ```conf env { job.name = "test_kafka_reader_cache_queue_size" parallelism = 5 job.mode = "STREAMING" checkpoint.interval = 90000 } source { Kafka { start_mode = "timestamp" start_mode.timestamp = 1779688800000 format = "json" format_error_handle_way = "skip" } } sink { Console { } } ``` ### Running Command ```shell $SEATUNNEL_HOME/bin/seatunnel.sh --config kafka_test.conf ``` ### Error Exception ```log 2026-05-26 11:34:19,617 ERROR [.s.c.s.r.f.SplitFetcherManager] [Source Data Fetcher for BlockingWorker-TaskGroupLocation{jobId=1111496543392825449, pipelineId=1, taskGroupId=5}] - Received uncaught exception. java.lang.OutOfMemoryError: Java heap space java.lang.RuntimeException: One or more fetchers have encountered exception at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:147) ~[seatunnel-transforms-v2.jar:2.3.13] at org.apache.seatunnel.connectors.seatunnel.common.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:167) ~[seatunnel-transforms-v2.jar:2.3.13] at org.apache.seatunnel.connectors.seatunnel.common.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:93) ~[seatunnel-transforms-v2.jar:2.3.13] at org.apache.seatunnel.engine.server.task.flow.SourceFlowLifeCycle.collect(SourceFlowLifeCycle.java:159) ~[seatunnel-starter.jar:2.3.13] at org.apache.seatunnel.engine.server.task.SourceSeaTunnelTask.collect(SourceSeaTunnelTask.java:127) ~[seatunnel-starter.jar:2.3.13] at org.apache.seatunnel.engine.server.task.SeaTunnelTask.stateProcess(SeaTunnelTask.java:165) ~[seatunnel-starter.jar:2.3.13] at org.apache.seatunnel.engine.server.task.SourceSeaTunnelTask.call(SourceSeaTunnelTask.java:132) ~[seatunnel-starter.jar:2.3.13] at org.apache.seatunnel.engine.server.TaskExecutionService$BlockingWorker.run(TaskExecutionService.java:683) ~[seatunnel-starter.jar:2.3.13] at org.apache.seatunnel.engine.server.TaskExecutionService$NamedTaskWrapper.run(TaskExecutionService.java:1012) ~[seatunnel-starter.jar:2.3.13] at org.apache.seatunnel.api.tracing.MDCRunnable.run(MDCRunnable.java:43) ~[seatunnel-starter.jar:2.3.13] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_342] at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[?:1.8.0_342] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_342] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_342] at java.lang.Thread.run(Thread.java:750) [?:1.8.0_342] Caused by: java.lang.OutOfMemoryError: Java heap space 2026-05-26 11:34:21,240 ERROR [o.a.k.c.c.KafkaConsumer ] [Source Data Fetcher for BlockingWorker-TaskGroupLocation{jobId=1111496543392825449, pipelineId=1, taskGroupId=5}] - [Consumer clientId=seatunnel-consumer-3, groupId=model_seatunnel] Failed to close coordinator java.lang.OutOfMemoryError: Java heap space 2026-05-26 11:34:22,319 ERROR [.s.c.s.r.f.SplitFetcherManager] [Source Data Fetcher for BlockingWorker-TaskGroupLocation{jobId=1111496543392825449, pipelineId=1, taskGroupId=5}] - Received uncaught exception. org.apache.kafka.common.KafkaException: Failed to close kafka consumer at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2431) ~[?:?] at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2379) ~[?:?] at org.apache.kafka.clients.consumer.KafkaConsumer.close(KafkaConsumer.java:2352) ~[?:?] at org.apache.seatunnel.connectors.seatunnel.kafka.source.KafkaPartitionSplitReader.close(KafkaPartitionSplitReader.java:300) ~[?:?] at org.apache.seatunnel.connectors.seatunnel.common.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:88) ~[seatunnel-transforms-v2.jar:2.3.13] at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) ~[?:1.8.0_342] at java.util.concurrent.FutureTask.run(FutureTask.java:266) ~[?:1.8.0_342] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) ~[?:1.8.0_342] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) ~[?:1.8.0_342] at java.lang.Thread.run(Thread.java:750) [?:1.8.0_342] Caused by: java.lang.OutOfMemoryError: Java heap space ``` ### Zeta or Flink or Spark Version _No response_ ### Java or Scala Version java1.8 ### Screenshots _No response_ ### Are you willing to submit PR? - [x] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- 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]
