This is an automated email from the ASF dual-hosted git repository.
bowenli86 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 90d05d57 [FLINK-38919][docs] Document PyFlink per-cluster offsets
(#280)
90d05d57 is described below
commit 90d05d5782317b662d4be80e7e55fc50b6e65b89
Author: bowenli86 <[email protected]>
AuthorDate: Mon Jul 6 15:49:58 2026 -0700
[FLINK-38919][docs] Document PyFlink per-cluster offsets (#280)
Documents the PyFlink per-cluster starting offset initializer API for
DynamicKafkaSource, following the merged FLINK-38918 implementation
apache/flink#28512 in apache/flink repo
---
.../docs/connectors/datastream/dynamic-kafka.md | 20 ++++++++++++++++++++
.../docs/connectors/datastream/dynamic-kafka.md | 20 ++++++++++++++++++++
2 files changed, 40 insertions(+)
diff --git a/docs/content.zh/docs/connectors/datastream/dynamic-kafka.md
b/docs/content.zh/docs/connectors/datastream/dynamic-kafka.md
index f9f17f3d..c7e033e0 100644
--- a/docs/content.zh/docs/connectors/datastream/dynamic-kafka.md
+++ b/docs/content.zh/docs/connectors/datastream/dynamic-kafka.md
@@ -139,6 +139,26 @@ DynamicKafkaSource<String> source =
.build();
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+starting_offsets = KafkaOffsetsInitializer.offsets({
+ KafkaTopicPartition("input-stream", 0): 100,
+ KafkaTopicPartition("input-stream", 1): 200,
+})
+
+metadata_service = SingleClusterTopicMetadataService(
+ "cluster-a",
+ {"bootstrap.servers": "localhost:9092"},
+ starting_offsets_initializer=starting_offsets)
+
+source = DynamicKafkaSource.builder() \
+ .set_kafka_metadata_service(metadata_service) \
+ .set_stream_ids({"input-stream"}) \
+ .set_starting_offsets(KafkaOffsetsInitializer.latest()) \
+ .set_value_only_deserializer(SimpleStringSchema()) \
+ .build()
+```
+{{< /tab >}}
{{< /tabs >}}
### Watermark 对齐
diff --git a/docs/content/docs/connectors/datastream/dynamic-kafka.md
b/docs/content/docs/connectors/datastream/dynamic-kafka.md
index a5488b51..6ef0454b 100644
--- a/docs/content/docs/connectors/datastream/dynamic-kafka.md
+++ b/docs/content/docs/connectors/datastream/dynamic-kafka.md
@@ -142,6 +142,26 @@ DynamicKafkaSource<String> source =
.build();
```
{{< /tab >}}
+{{< tab "Python" >}}
+```python
+starting_offsets = KafkaOffsetsInitializer.offsets({
+ KafkaTopicPartition("input-stream", 0): 100,
+ KafkaTopicPartition("input-stream", 1): 200,
+})
+
+metadata_service = SingleClusterTopicMetadataService(
+ "cluster-a",
+ {"bootstrap.servers": "localhost:9092"},
+ starting_offsets_initializer=starting_offsets)
+
+source = DynamicKafkaSource.builder() \
+ .set_kafka_metadata_service(metadata_service) \
+ .set_stream_ids({"input-stream"}) \
+ .set_starting_offsets(KafkaOffsetsInitializer.latest()) \
+ .set_value_only_deserializer(SimpleStringSchema()) \
+ .build()
+```
+{{< /tab >}}
{{< /tabs >}}
### Split Assignment Mode
