This is an automated email from the ASF dual-hosted git repository.
davidzollo pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 6736f57e99 [Fix][Kafka] Reduce default reader_cache_queue_size from
1024 to 2 to prevent OOM (#10954)
6736f57e99 is described below
commit 6736f57e99a4bcb4b2098b6b0690564e4ed7322b
Author: zhiwei.niu <[email protected]>
AuthorDate: Wed May 27 11:17:34 2026 +0800
[Fix][Kafka] Reduce default reader_cache_queue_size from 1024 to 2 to
prevent OOM (#10954)
---
docs/en/connectors/source/Kafka.md | 19 ++++++++++++++++++-
docs/zh/connectors/source/Kafka.md | 19 ++++++++++++++++++-
.../seatunnel/kafka/config/KafkaSourceOptions.java | 7 +++++--
3 files changed, 41 insertions(+), 4 deletions(-)
diff --git a/docs/en/connectors/source/Kafka.md
b/docs/en/connectors/source/Kafka.md
index a43032b5a3..c66e197cce 100644
--- a/docs/en/connectors/source/Kafka.md
+++ b/docs/en/connectors/source/Kafka.md
@@ -59,13 +59,30 @@ They can be downloaded via install-plugin.sh or from the
Maven central repositor
| protobuf_message_name | String
| No | - |
Effective when the format is set to protobuf, specifies the Message name
[...]
| protobuf_schema | String
| No | - |
Effective when the format is set to protobuf, specifies the Schema definition
[...]
| strip_schema_registry_header | Boolean
| No | false |
Effective when the format is set to protobuf. Whether to strip the Confluent
Schema Registry wire format header (magic byte, schema id and message indexes)
before protobuf deserialization. This option is useful when consuming Protobuf
messages that were encoded using Confluent Schema Registry. When enabled, the
connector will try to de [...]
-| reader_cache_queue_size | Integer
| No | 1024 |
The reader shard cache queue is used to cache the data corresponding to the
shards. The size of the shard cache depends on the number of shards obtained by
each reader, rather than the amount of data in each shard.
[...]
+| reader_cache_queue_size | Integer
| No | 2 |
The capacity of the fetcher-to-reader element queue. Each element is one
`consumer.poll()` batch, not a single message. See
[reader_cache_queue_size](#reader_cache_queue_size) for details. |
| is_native | Boolean
| No | false |
Supports retaining the source information of the record.
> On restore from checkpoint or savepoint, Kafka Source resumes from the
> checkpointed split offsets.
> `start_mode` and consumer-group offsets are only used for the first startup
> or for newly
> discovered partitions that do not have checkpointed state yet.
+### reader_cache_queue_size
+
+The maximum number of poll-result batches that the connector buffers between
the fetcher thread and the reader thread.
+
+:::tip
+
+Each element in the queue is one complete `consumer.poll()` result, which may
contain up to `max.poll.records` (default 500) messages.
+When back-pressure occurs downstream, the queue may fill up, and the upper
bound of messages held in memory is `reader_cache_queue_size ×
max.poll.records`.
+
+:::
+
+:::caution
+
+When consuming large messages, a high value can lead to excessive heap usage.
If you observe memory pressure, reduce this value or lower
`kafka.max.poll.records`.
+
+:::
+
### debezium_record_table_filter
We can use `debezium_record_table_filter` to filter the data in the debezium
format. The configuration is as follows:
diff --git a/docs/zh/connectors/source/Kafka.md
b/docs/zh/connectors/source/Kafka.md
index 78d7f41668..90cef2a8dd 100644
--- a/docs/zh/connectors/source/Kafka.md
+++ b/docs/zh/connectors/source/Kafka.md
@@ -59,12 +59,29 @@ import ChangeLog from '../changelog/connector-kafka.md';
| protobuf_message_name | String |
否 | - | 当格式设置为 protobuf 时有效，指定消息名称。
|
| protobuf_schema | String |
否 | - | 当格式设置为 protobuf 时有效，指定 Schema 定义。
|
| strip_schema_registry_header | Boolean |
否 | false | 当格式设置为 protobuf 时有效。是否在 Protobuf 反序列化之前去除
Confluent Schema Registry 线格式头部（magic byte、schema id 和 message indexes）。当消费使用
Confluent Schema Registry 编码的 Protobuf 消息时，此选项非常有用。启用后，连接器将尝试在解析 Protobuf
消息之前检测并删除 Schema Registry 头部。如果未检测到头部，它将回退到标准的 Protobuf 反序列化。
[...]
-| reader_cache_queue_size | Integer |
否 | 1024 |
Reader分片缓存队列，用于缓存分片对应的数据。占用大小取决于每个reader得到的分片量，而不是每个分片的数据量。
|
+| reader_cache_queue_size | Integer |
否 | 2 | Fetcher 与 Reader 线程之间缓冲队列的容量。每个元素是一次
`consumer.poll()` 的整批结果，而非单条消息。详见
[reader_cache_queue_size](#reader_cache_queue_size)。 |
| is_native | Boolean |
No | false | 支持保留record的源信息。
|
> 从 checkpoint 或 savepoint 恢复时，Kafka Source 会优先使用 checkpoint 中保存的 split offset。
> `start_mode` 和 consumer group offset 只在首次启动，或为尚未存在 checkpoint
> 状态的新发现分区初始化位点时生效。
+### reader_cache_queue_size
+
+连接器在 Fetcher 线程和 Reader 线程之间缓冲的 poll 结果批次的最大数量。
+
+:::tip
+
+队列中的每个元素是一次完整的 `consumer.poll()` 结果，最多包含 `max.poll.records`（默认 500）条消息。
+当下游产生背压时，队列可能被填满，此时驻留在内存中的消息数上限为 `reader_cache_queue_size × max.poll.records`。
+
+:::
+
+:::caution
+
+当消费的消息体较大时，过高的值会导致堆内存占用过高。如果观察到内存压力，请减小此值或降低 `kafka.max.poll.records`。
+
+:::
+
### debezium_record_table_filter
我们可以使用 `debezium_record_table_filter` 来过滤 debezium 格式的数据。配置如下：
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/config/KafkaSourceOptions.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/config/KafkaSourceOptions.java
index 33e8413cef..e351c7e26b 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/config/KafkaSourceOptions.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/config/KafkaSourceOptions.java
@@ -44,8 +44,11 @@ public class KafkaSourceOptions extends KafkaBaseOptions {
public static final Option<Integer> READER_CACHE_QUEUE_SIZE =
Options.key("reader_cache_queue_size")
.intType()
- .defaultValue(1024)
- .withDescription("The size of reader queue.");
+ .defaultValue(2)
+ .withDescription(
+ "The capacity of the fetcher-to-reader element
queue. "
+ + "Each element is one consumer.poll()
batch, not a single message. "
+ + "Actual buffered messages ≈ this value ×
max.poll.records.");
public static final Option<Boolean> COMMIT_ON_CHECKPOINT =
Options.key("commit_on_checkpoint")
