This is an automated email from the ASF dual-hosted git repository.
1996fanrui pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
The following commit(s) were added to refs/heads/main by this push:
new 066daaee [FLINK-39705] Remove Zookeeper references
066daaee is described below
commit 066daaee72288fa66f837f559ec14712e6040ec2
Author: Aleksandr Savonin <[email protected]>
AuthorDate: Wed May 20 12:40:45 2026 +0200
[FLINK-39705] Remove Zookeeper references
---
docs/content.zh/docs/connectors/table/kafka.md | 6 +++---
docs/content/docs/connectors/table/kafka.md | 6 +++---
.../apache/flink/streaming/connectors/kafka/config/BoundedMode.java | 4 ++--
.../apache/flink/streaming/connectors/kafka/config/StartupMode.java | 4 +---
.../streaming/connectors/kafka/table/KafkaConnectorOptions.java | 4 ++--
.../java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java | 3 ---
.../flink/streaming/connectors/kafka/table/KafkaTableTestBase.java | 3 ---
7 files changed, 11 insertions(+), 19 deletions(-)
diff --git a/docs/content.zh/docs/connectors/table/kafka.md
b/docs/content.zh/docs/connectors/table/kafka.md
index 9df680df..8eba82f1 100644
--- a/docs/content.zh/docs/connectors/table/kafka.md
+++ b/docs/content.zh/docs/connectors/table/kafka.md
@@ -506,13 +506,13 @@ ROW<`version` INT, `behavior` STRING>
`scan.startup.mode` 配置项决定了 Kafka consumer 的启动模式。有效值为：
-* `group-offsets`：从 Zookeeper/Kafka 中某个指定的消费组已提交的偏移量开始。
+* `group-offsets`：从 Kafka 中某个指定的消费组已提交的偏移量开始。
* `earliest-offset`：从可能的最早偏移量开始。
* `latest-offset`：从最末尾偏移量开始。
* `timestamp`：从用户为每个 partition 指定的时间戳开始。
* `specific-offsets`：从用户为每个 partition 指定的偏移量开始。
-默认值 `group-offsets` 表示从 Zookeeper/Kafka 中最近一次已提交的偏移量开始消费。
+默认值 `group-offsets` 表示从 Kafka 中最近一次已提交的偏移量开始消费。
如果使用了 `timestamp`，必须使用另外一个配置项 `scan.startup.timestamp-millis` 来指定一个从格林尼治标准时间
1970 年 1 月 1 日 00:00:00.000 开始计算的毫秒单位时间戳作为起始时间。
@@ -523,7 +523,7 @@ ROW<`version` INT, `behavior` STRING>
The config option `scan.bounded.mode` specifies the bounded mode for Kafka
consumer. The valid enumerations are:
<ul>
-<li><span markdown="span">`group-offsets`</span>: bounded by committed offsets
in ZooKeeper / Kafka brokers of a specific consumer group. This is evaluated at
the start of consumption from a given partition.</li>
+<li><span markdown="span">`group-offsets`</span>: bounded by committed offsets
in Kafka brokers of a specific consumer group. This is evaluated at the start
of consumption from a given partition.</li>
<li><span markdown="span">`latest-offset`</span>: bounded by latest offsets.
This is evaluated at the start of consumption from a given partition.</li>
<li><span markdown="span">`timestamp`</span>: bounded by a user-supplied
timestamp.</li>
<li><span markdown="span">`specific-offsets`</span>: bounded by user-supplied
specific offsets for each partition.</li>
diff --git a/docs/content/docs/connectors/table/kafka.md
b/docs/content/docs/connectors/table/kafka.md
index 12b0821c..fab543cd 100644
--- a/docs/content/docs/connectors/table/kafka.md
+++ b/docs/content/docs/connectors/table/kafka.md
@@ -557,13 +557,13 @@ Note that topic list and topic pattern only work in
sources. In sinks, Flink cur
The config option `scan.startup.mode` specifies the startup mode for Kafka
consumer. The valid enumerations are:
-* `group-offsets`: start from committed offsets in ZK / Kafka brokers of a
specific consumer group.
+* `group-offsets`: start from committed offsets in Kafka brokers of a specific
consumer group.
* `earliest-offset`: start from the earliest offset possible.
* `latest-offset`: start from the latest offset.
* `timestamp`: start from user-supplied timestamp for each partition.
* `specific-offsets`: start from user-supplied specific offsets for each
partition.
-The default option value is `group-offsets` which indicates to consume from
last committed offsets in ZK / Kafka brokers.
+The default option value is `group-offsets` which indicates to consume from
last committed offsets in Kafka brokers.
If `timestamp` is specified, another config option
`scan.startup.timestamp-millis` is required to specify a specific startup
timestamp in milliseconds since January 1, 1970 00:00:00.000 GMT.
@@ -574,7 +574,7 @@ e.g. an option value
`partition:0,offset:42;partition:1,offset:300` indicates of
The config option `scan.bounded.mode` specifies the bounded mode for Kafka
consumer. The valid enumerations are:
<ul>
-<li><span markdown="span">`group-offsets`</span>: bounded by committed offsets
in ZooKeeper / Kafka brokers of a specific consumer group. This is evaluated at
the start of consumption from a given partition.</li>
+<li><span markdown="span">`group-offsets`</span>: bounded by committed offsets
in Kafka brokers of a specific consumer group. This is evaluated at the start
of consumption from a given partition.</li>
<li><span markdown="span">`latest-offset`</span>: bounded by latest offsets.
This is evaluated at the start of consumption from a given partition.</li>
<li><span markdown="span">`timestamp`</span>: bounded by a user-supplied
timestamp.</li>
<li><span markdown="span">`specific-offsets`</span>: bounded by user-supplied
specific offsets for each partition.</li>
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/config/BoundedMode.java
b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/config/BoundedMode.java
index beb2306b..cb1993db 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/config/BoundedMode.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/config/BoundedMode.java
@@ -27,8 +27,8 @@ public enum BoundedMode {
UNBOUNDED,
/**
- * End from committed offsets in ZK / Kafka brokers of a specific consumer
group. This is
- * evaluated at the start of consumption from a given partition.
+ * End from committed offsets in Kafka brokers of a specific consumer
group. This is evaluated
+ * at the start of consumption from a given partition.
*/
GROUP_OFFSETS,
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/config/StartupMode.java
b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/config/StartupMode.java
index 0aa43ecb..e13b657e 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/config/StartupMode.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/config/StartupMode.java
@@ -24,9 +24,7 @@ import
org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition
@Internal
public enum StartupMode {
- /**
- * Start from committed offsets in ZK / Kafka brokers of a specific
consumer group (default).
- */
+ /** Start from committed offsets in Kafka brokers of a specific consumer
group (default). */
GROUP_OFFSETS(KafkaTopicPartitionStateSentinel.GROUP_OFFSET),
/** Start from the earliest offset possible. */
diff --git
a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaConnectorOptions.java
b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaConnectorOptions.java
index 0639e812..2dc937ee 100644
---
a/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaConnectorOptions.java
+++
b/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaConnectorOptions.java
@@ -314,7 +314,7 @@ public class KafkaConnectorOptions {
GROUP_OFFSETS(
"group-offsets",
text(
- "Start from committed offsets in ZooKeeper / Kafka
brokers of a specific consumer group.")),
+ "Start from committed offsets in Kafka brokers of a
specific consumer group.")),
TIMESTAMP("timestamp", text("Start from user-supplied timestamp for
each partition.")),
SPECIFIC_OFFSETS(
"specific-offsets",
@@ -350,7 +350,7 @@ public class KafkaConnectorOptions {
GROUP_OFFSETS(
"group-offsets",
text(
- "Bounded by committed offsets in ZooKeeper / Kafka
brokers of a specific"
+ "Bounded by committed offsets in Kafka brokers of a
specific"
+ " consumer group. This is evaluated at the
start of consumption"
+ " from a given partition.")),
TIMESTAMP("timestamp", text("Bounded by a user-supplied timestamp.")),
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
index 19ab1c6a..6e382e06 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
@@ -134,7 +134,6 @@ class KafkaSinkITCase {
private static final Logger LOG =
LoggerFactory.getLogger(KafkaSinkITCase.class);
private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
private static final Network NETWORK = Network.newNetwork();
- private static final int ZK_TIMEOUT_MILLIS = 30000;
private static final short TOPIC_REPLICATION_FACTOR = 1;
private static final long CHECKPOINT_PATH_LOOKUP_TIMEOUT_SECONDS = 10;
private static final long CHECKPOINT_PATH_LOOKUP_POLL_INTERVAL_MILLIS =
200;
@@ -762,8 +761,6 @@ class KafkaSinkITCase {
standardProps.put("enable.auto.commit", false);
standardProps.put("auto.offset.reset", "earliest");
standardProps.put("max.partition.fetch.bytes", 256);
- standardProps.put("zookeeper.session.timeout.ms", ZK_TIMEOUT_MILLIS);
- standardProps.put("zookeeper.connection.timeout.ms",
ZK_TIMEOUT_MILLIS);
return standardProps;
}
diff --git
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java
index 85e20cab..5db67f6a 100644
---
a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java
+++
b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/table/KafkaTableTestBase.java
@@ -63,7 +63,6 @@ abstract class KafkaTableTestBase extends AbstractTestBase {
private static final Logger LOG =
LoggerFactory.getLogger(KafkaTableTestBase.class);
private static final String INTER_CONTAINER_KAFKA_ALIAS = "kafka";
- private static final int zkTimeoutMills = 30000;
@Container
public static final TestKafkaContainer KAFKA_CONTAINER =
@@ -115,8 +114,6 @@ abstract class KafkaTableTestBase extends AbstractTestBase {
standardProps.put("enable.auto.commit", false);
standardProps.put("auto.offset.reset", "earliest");
standardProps.put("max.partition.fetch.bytes", 256);
- standardProps.put("zookeeper.session.timeout.ms", zkTimeoutMills);
- standardProps.put("zookeeper.connection.timeout.ms", zkTimeoutMills);
return standardProps;
}
