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 7001ed6a3b [Fix][Connector-V2][Kafka] Fixed: inverted
commit_on_checkpoint for… (#10320)
7001ed6a3b is described below
commit 7001ed6a3b696fb8b2022840e66d63c7e436c76c
Author: tomma <[email protected]>
AuthorDate: Sun Jul 19 09:51:18 2026 +0800
[Fix][Connector-V2][Kafka] Fixed: inverted commit_on_checkpoint for…
(#10320)
Co-authored-by: DanielLeens <[email protected]>
---
docs/en/connectors/source/Kafka.md | 2 +-
docs/zh/connectors/source/Kafka.md | 2 +-
.../seatunnel/kafka/config/KafkaSourceOptions.java | 2 +-
.../kafka/source/KafkaPartitionSplitReader.java | 74 +++++++++++--------
.../source/KafkaPartitionSplitReaderTest.java | 86 ++++++++++++++++++++++
5 files changed, 131 insertions(+), 35 deletions(-)
diff --git a/docs/en/connectors/source/Kafka.md
b/docs/en/connectors/source/Kafka.md
index 28e23bfef9..7402b4e6c2 100644
--- a/docs/en/connectors/source/Kafka.md
+++ b/docs/en/connectors/source/Kafka.md
@@ -42,7 +42,7 @@ They can be downloaded via install-plugin.sh or from the
Maven central repositor
| bootstrap.servers | String
| Yes | - |
Comma separated list of Kafka brokers.
[...]
| pattern | Boolean
| No | false | If
`pattern` is set to `true`,the regular expression for a pattern of topic names
to read from. All topics in clients with names that match the specified regular
expression will be subscribed by the consumer.
[...]
| consumer.group | String
| No | SeaTunnel-Consumer-Group |
`Kafka consumer group id`, used to distinguish different consumer groups.
[...]
-| commit_on_checkpoint | Boolean
| No | true | If
true the consumer's offset will be periodically committed in the background.
[...]
+| commit_on_checkpoint | Boolean
| No | true | If
true, consumer offsets are committed only after a SeaTunnel checkpoint
completes, and Kafka auto commit is disabled. If false, checkpoint commits are
disabled and Kafka auto commit is enabled.
[...]
| poll.timeout | Long
| No | 10000 | The
interval(millis) for poll messages.
[...]
| kafka.config | Map
| No | - | In
addition to the above necessary parameters that must be specified by the `Kafka
consumer` client, users can also specify multiple `consumer` client
non-mandatory parameters, covering [all consumer parameters specified in the
official Kafka
document](https://kafka.apache.org/documentation.html#consumerconfigs).
[...]
| schema | Config
| No | - | The
structure of the data, including field names and field types. For more details,
please refer to [Schema
Feature](../../introduction/concepts/schema-feature.md).
[...]
diff --git a/docs/zh/connectors/source/Kafka.md
b/docs/zh/connectors/source/Kafka.md
index 89921afaa2..89dfd2da13 100644
--- a/docs/zh/connectors/source/Kafka.md
+++ b/docs/zh/connectors/source/Kafka.md
@@ -42,7 +42,7 @@ import ChangeLog from '../changelog/connector-kafka.md';
| bootstrap.servers | String |
是 | - | 逗号分隔的 Kafka brokers 列表。
|
| pattern | Boolean |
否 | false | 如果 `pattern` 设置为
`true`，则会使用指定的正则表达式匹配并订阅主题。
|
| consumer.group | String |
否 | SeaTunnel-Consumer-Group | `Kafka 消费者组 ID`，用于区分不同的消费者组。
|
-| commit_on_checkpoint | Boolean |
否 | true | 如果为 true，消费者的偏移量将会定期在后台提交。
|
+| commit_on_checkpoint | Boolean |
否 | true | 如果为 true，仅在 SeaTunnel checkpoint
完成后提交消费者偏移量，并禁用 Kafka 自动提交；如果为 false，则禁用 checkpoint 提交并启用 Kafka 自动提交。
|
| poll.timeout | Long |
否 | 10000 | kafka主动拉取时间间隔(毫秒)。
|
| kafka.config | Map |
否 | - | 除了上述必要参数外，用户还可以指定多个非强制的消费者客户端参数，覆盖 [Kafka
官方文档](https://kafka.apache.org/documentation.html#consumerconfigs) 中指定的所有消费者参数。
|
| schema | Config |
否 | - | 数据结构，包括字段名称和字段类型。更多详情请参考 [Schema
特性](../../introduction/concepts/schema-feature.md)。
|
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 9373290978..f89ee28cc6 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
@@ -55,7 +55,7 @@ public class KafkaSourceOptions extends KafkaBaseOptions {
.booleanType()
.defaultValue(true)
.withDescription(
- "If true the consumer's offset will be
periodically committed in the background.");
+ "If true, consumer offsets are committed only
after a SeaTunnel checkpoint completes, and Kafka auto commit is disabled. If
false, checkpoint commits are disabled and Kafka auto commit is enabled.");
public static final Option<Boolean> DEBEZIUM_RECORD_INCLUDE_SCHEMA =
Options.key("debezium_record_include_schema")
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReader.java
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReader.java
index 9609534090..a61fadc65a 100644
---
a/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReader.java
+++
b/seatunnel-connectors-v2/connector-kafka/src/main/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReader.java
@@ -307,43 +307,53 @@ public class KafkaPartitionSplitReader
consumer.commitAsync(offsetsToCommit, offsetCommitCallback);
}
+ /**
+ * Builds Kafka consumer properties for the source reader.
+ *
+ * <p>The source either relies on checkpoint-triggered commits or Kafka
auto commit, but never
+ * both at the same time.
+ */
+ static Properties buildConsumerProperties(
+ KafkaSourceConfig kafkaSourceConfig, int subtaskId, String
clientIdPrefix) {
+ Properties props = new Properties();
+ kafkaSourceConfig
+ .getProperties()
+ .forEach(
+ (key, value) ->
+ props.setProperty(String.valueOf(key),
String.valueOf(value)));
+ props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,
kafkaSourceConfig.getConsumerGroup());
+ props.setProperty(
+ ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaSourceConfig.getBootstrap());
+ if (kafkaSourceConfig.getProperties().get("client.id") == null) {
+ props.setProperty(
+ ConsumerConfig.CLIENT_ID_CONFIG, clientIdPrefix +
"-consumer-" + subtaskId);
+ } else {
+ props.setProperty(
+ ConsumerConfig.CLIENT_ID_CONFIG,
+
kafkaSourceConfig.getProperties().get("client.id").toString()
+ + "-"
+ + subtaskId);
+ }
+ props.setProperty(
+ ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+ ByteArrayDeserializer.class.getName());
+ props.setProperty(
+ ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+ ByteArrayDeserializer.class.getName());
+ props.setProperty(
+ ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
+ String.valueOf(!kafkaSourceConfig.isCommitOnCheckpoint()));
+ props.setProperty(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG,
"false");
+ return props;
+ }
+
private KafkaConsumer<byte[], byte[]> initConsumer(
KafkaSourceConfig kafkaSourceConfig, int subtaskId) {
try (TemporaryClassLoaderContext ignored =
TemporaryClassLoaderContext.of(kafkaSourceConfig.getClass().getClassLoader())) {
- Properties props = new Properties();
- kafkaSourceConfig
- .getProperties()
- .forEach(
- (key, value) ->
- props.setProperty(String.valueOf(key),
String.valueOf(value)));
- props.setProperty(ConsumerConfig.GROUP_ID_CONFIG,
kafkaSourceConfig.getConsumerGroup());
- props.setProperty(
- ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
kafkaSourceConfig.getBootstrap());
- if (this.kafkaSourceConfig.getProperties().get("client.id") ==
null) {
- props.setProperty(
- ConsumerConfig.CLIENT_ID_CONFIG,
- CLIENT_ID_PREFIX + "-consumer-" + subtaskId);
- } else {
- props.setProperty(
- ConsumerConfig.CLIENT_ID_CONFIG,
-
this.kafkaSourceConfig.getProperties().get("client.id").toString()
- + "-"
- + subtaskId);
- }
- props.setProperty(
- ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
- ByteArrayDeserializer.class.getName());
- props.setProperty(
- ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
- ByteArrayDeserializer.class.getName());
- props.setProperty(
- ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG,
- String.valueOf(kafkaSourceConfig.isCommitOnCheckpoint()));
-
- // Disable auto create topics feature
- props.setProperty(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG,
"false");
+ Properties props =
+ buildConsumerProperties(kafkaSourceConfig, subtaskId,
CLIENT_ID_PREFIX);
return new KafkaConsumer<>(props);
}
}
diff --git
a/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReaderTest.java
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReaderTest.java
new file mode 100644
index 0000000000..e8d7890a6a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-kafka/src/test/java/org/apache/seatunnel/connectors/seatunnel/kafka/source/KafkaPartitionSplitReaderTest.java
@@ -0,0 +1,86 @@
+/*
+ * 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.seatunnel.connectors.seatunnel.kafka.source;
+
+import org.apache.seatunnel.api.configuration.ReadonlyConfig;
+
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Properties;
+
+/**
+ * Tests the Kafka source reader consumer property mapping.
+ *
+ * <p>This guards the offset commit contract between SeaTunnel checkpoint
commits and Kafka auto
+ * commit.
+ */
+public class KafkaPartitionSplitReaderTest {
+
+ /** Verifies that checkpoint commits disable Kafka auto commit. */
+ @Test
+ void
testBuildConsumerPropertiesDisablesAutoCommitWhenCheckpointCommitEnabled() {
+ KafkaSourceConfig kafkaSourceConfig = new
KafkaSourceConfig(createConfig(true));
+
+ Properties properties =
+ KafkaPartitionSplitReader.buildConsumerProperties(
+ kafkaSourceConfig, 1, "seatunnel-test");
+
+ Assertions.assertEquals(
+ "false",
properties.getProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
+ }
+
+ /** Verifies that Kafka auto commit stays enabled when checkpoint commits
are disabled. */
+ @Test
+ void
testBuildConsumerPropertiesEnablesAutoCommitWhenCheckpointCommitDisabled() {
+ KafkaSourceConfig kafkaSourceConfig = new
KafkaSourceConfig(createConfig(false));
+
+ Properties properties =
+ KafkaPartitionSplitReader.buildConsumerProperties(
+ kafkaSourceConfig, 2, "seatunnel-test");
+
+ Assertions.assertEquals(
+ "true",
properties.getProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG));
+ }
+
+ /** Creates the minimum Kafka source config used by the tests. */
+ private ReadonlyConfig createConfig(boolean commitOnCheckpoint) {
+ Map<String, Object> configMap = new HashMap<>();
+ configMap.put("bootstrap.servers", "localhost:9092");
+ configMap.put("group.id", "test-group");
+ configMap.put("topic", "test-topic");
+ configMap.put("schema", createSchema());
+ configMap.put("format", "text");
+ configMap.put("commit_on_checkpoint", commitOnCheckpoint);
+ return ReadonlyConfig.fromMap(configMap);
+ }
+
+ /** Creates a minimal text schema for constructing {@link
KafkaSourceConfig}. */
+ private Map<String, Object> createSchema() {
+ Map<String, Object> schemaFields = new HashMap<>();
+ schemaFields.put("id", "int");
+
+ Map<String, Object> schema = new HashMap<>();
+ schema.put("fields", schemaFields);
+ return schema;
+ }
+}
