From 3fcab7da9e21d63ca95a01d8026c0a044c426083 Mon Sep 17 00:00:00 2001 From: Ashfaq <105435085+Ashfaqbs@users.noreply.github.com> Date: Sun, 18 May 2025 10:25:59 +0530 Subject: [PATCH 01/10] Add Kafka to Kafka RAW example pipeline configuration --- .../yaml/examples/kafka-to-kafka.yaml | 22 +++++++++++++++++++ 1 file changed, 22 insertions(+) create mode 100644 sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml diff --git a/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml b/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml new file mode 100644 index 000000000000..802bad1952e5 --- /dev/null +++ b/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml @@ -0,0 +1,22 @@ +pipeline: + type: chain + transforms: + - type: ReadFromKafka + name: KafkaSource + config: + topic: "i-topic" + bootstrap_servers: "localhost:9092" + format: RAW + consumer_config: + group.id: beam-mirror-001 + auto.offset.reset: earliest + + - type: WriteToKafka + name: MirrorToNTopic + config: + topic: "o-topic" + bootstrap_servers: "localhost:9092" + format: RAW + producer_config_update: + key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer + value.serializer: org.apache.kafka.common.serialization.ByteArraySerializer \ No newline at end of file From bb56d4f50f5a463cd739a7f24a82245c1212a88a Mon Sep 17 00:00:00 2001 From: Ashfaq <105435085+Ashfaqbs@users.noreply.github.com> Date: Sun, 29 Jun 2025 11:54:00 +0530 Subject: [PATCH 02/10] Add license header to kafka-to-kafka.yaml --- .../yaml/examples/kafka-to-kafka.yaml | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml b/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml index 802bad1952e5..52d6108035a8 100644 --- a/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml +++ b/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml @@ -1,3 +1,20 @@ +# 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. + + + pipeline: type: chain transforms: From c8be26fb2384cf8a15b289958012d50b2c1e2901 Mon Sep 17 00:00:00 2001 From: Ashfaq <105435085+Ashfaqbs@users.noreply.github.com> Date: Sun, 29 Jun 2025 11:59:17 +0530 Subject: [PATCH 03/10] Add description to kafka-to-kafka.yaml for clarity on pipeline functionality --- sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml b/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml index 52d6108035a8..5551884145a1 100644 --- a/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml +++ b/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml @@ -13,6 +13,11 @@ # See the License for the specific language governing permissions and # limitations under the License. +# Description: +# This Apache Beam YAML pipeline demonstrates a basic Kafka-to-Kafka mirroring example. +# It reads raw byte messages from a source Kafka topic and writes them to a target Kafka topic. +# No transformation or decoding is applied. +# Useful for validating Kafka I/O setup using Beam YAML DSL. pipeline: From 5ff0adcf8a6d7038a271fe594da5befa3cd18ee6 Mon Sep 17 00:00:00 2001 From: Ashfaq <105435085+Ashfaqbs@users.noreply.github.com> Date: Sun, 29 Jun 2025 12:21:20 +0530 Subject: [PATCH 04/10] Remove kafka-to-kafka.yaml example pipeline configuration --- .../yaml/examples/kafka-to-kafka.yaml | 44 ------------------- 1 file changed, 44 deletions(-) delete mode 100644 sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml diff --git a/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml b/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml deleted file mode 100644 index 5551884145a1..000000000000 --- a/sdks/python/apache_beam/yaml/examples/kafka-to-kafka.yaml +++ /dev/null @@ -1,44 +0,0 @@ -# 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. - -# Description: -# This Apache Beam YAML pipeline demonstrates a basic Kafka-to-Kafka mirroring example. -# It reads raw byte messages from a source Kafka topic and writes them to a target Kafka topic. -# No transformation or decoding is applied. -# Useful for validating Kafka I/O setup using Beam YAML DSL. - - -pipeline: - type: chain - transforms: - - type: ReadFromKafka - name: KafkaSource - config: - topic: "i-topic" - bootstrap_servers: "localhost:9092" - format: RAW - consumer_config: - group.id: beam-mirror-001 - auto.offset.reset: earliest - - - type: WriteToKafka - name: MirrorToNTopic - config: - topic: "o-topic" - bootstrap_servers: "localhost:9092" - format: RAW - producer_config_update: - key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer - value.serializer: org.apache.kafka.common.serialization.ByteArraySerializer \ No newline at end of file From d8d6818d1b9a6d00095f28ebed68926023c76dad Mon Sep 17 00:00:00 2001 From: Ashfaq <105435085+Ashfaqbs@users.noreply.github.com> Date: Sun, 29 Jun 2025 12:21:35 +0530 Subject: [PATCH 05/10] Add kafka-to-kafka_example.yaml pipeline configuration --- .../yaml/examples/kafka-to-kafka_example.yaml | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) create mode 100644 sdks/python/apache_beam/yaml/examples/kafka-to-kafka_example.yaml diff --git a/sdks/python/apache_beam/yaml/examples/kafka-to-kafka_example.yaml b/sdks/python/apache_beam/yaml/examples/kafka-to-kafka_example.yaml new file mode 100644 index 000000000000..5551884145a1 --- /dev/null +++ b/sdks/python/apache_beam/yaml/examples/kafka-to-kafka_example.yaml @@ -0,0 +1,44 @@ +# 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. + +# Description: +# This Apache Beam YAML pipeline demonstrates a basic Kafka-to-Kafka mirroring example. +# It reads raw byte messages from a source Kafka topic and writes them to a target Kafka topic. +# No transformation or decoding is applied. +# Useful for validating Kafka I/O setup using Beam YAML DSL. + + +pipeline: + type: chain + transforms: + - type: ReadFromKafka + name: KafkaSource + config: + topic: "i-topic" + bootstrap_servers: "localhost:9092" + format: RAW + consumer_config: + group.id: beam-mirror-001 + auto.offset.reset: earliest + + - type: WriteToKafka + name: MirrorToNTopic + config: + topic: "o-topic" + bootstrap_servers: "localhost:9092" + format: RAW + producer_config_update: + key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer + value.serializer: org.apache.kafka.common.serialization.ByteArraySerializer \ No newline at end of file From 85057d12f6c6731b6946a7918bd94d4ca503d883 Mon Sep 17 00:00:00 2001 From: Ashfaq <105435085+Ashfaqbs@users.noreply.github.com> Date: Thu, 28 Aug 2025 22:55:43 +0530 Subject: [PATCH 06/10] Move to transforms/io; add hermetic tests; align Kafka YAML options; drop old example --- .../yaml/examples/kafka-to-kafka_example.yaml | 44 ---------- .../transforms/io/kafka_to_kafka.yaml | 84 +++++++++++++++++++ 2 files changed, 84 insertions(+), 44 deletions(-) delete mode 100644 sdks/python/apache_beam/yaml/examples/kafka-to-kafka_example.yaml create mode 100644 sdks/python/apache_beam/yaml/examples/transforms/io/kafka_to_kafka.yaml diff --git a/sdks/python/apache_beam/yaml/examples/kafka-to-kafka_example.yaml b/sdks/python/apache_beam/yaml/examples/kafka-to-kafka_example.yaml deleted file mode 100644 index 5551884145a1..000000000000 --- a/sdks/python/apache_beam/yaml/examples/kafka-to-kafka_example.yaml +++ /dev/null @@ -1,44 +0,0 @@ -# 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. - -# Description: -# This Apache Beam YAML pipeline demonstrates a basic Kafka-to-Kafka mirroring example. -# It reads raw byte messages from a source Kafka topic and writes them to a target Kafka topic. -# No transformation or decoding is applied. -# Useful for validating Kafka I/O setup using Beam YAML DSL. - - -pipeline: - type: chain - transforms: - - type: ReadFromKafka - name: KafkaSource - config: - topic: "i-topic" - bootstrap_servers: "localhost:9092" - format: RAW - consumer_config: - group.id: beam-mirror-001 - auto.offset.reset: earliest - - - type: WriteToKafka - name: MirrorToNTopic - config: - topic: "o-topic" - bootstrap_servers: "localhost:9092" - format: RAW - producer_config_update: - key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer - value.serializer: org.apache.kafka.common.serialization.ByteArraySerializer \ No newline at end of file diff --git a/sdks/python/apache_beam/yaml/examples/transforms/io/kafka_to_kafka.yaml b/sdks/python/apache_beam/yaml/examples/transforms/io/kafka_to_kafka.yaml new file mode 100644 index 000000000000..dd061375b121 --- /dev/null +++ b/sdks/python/apache_beam/yaml/examples/transforms/io/kafka_to_kafka.yaml @@ -0,0 +1,84 @@ +# 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. + +# Description: +# Kafka-to-Kafka RAW mirroring using Beam YAML. +# Reads bytes from a source topic and writes bytes to a target topic. +# Minimal pass-through, useful for validating Kafka I/O setup. +# Includes an inline test that mocks Kafka I/O for hermetic CI. + +pipeline: + type: chain + transforms: + - type: ReadFromKafka + name: KafkaSource + config: + format: RAW + topic: "{{ SOURCE_TOPIC | default('i-topic') }}" + bootstrap_servers: "{{ BOOTSTRAP_SERVERS | default('localhost:9092') }}" + auto_offset_reset_config: earliest + consumer_config: + group.id: beam-mirror-001 + + # Map RAW Kafka read to KV expected by WriteToKafka. + - type: MapToFields + name: ToKafkaKV + config: + language: python + fields: + key: + callable: | + def k(row): + try: + return row.key + except AttributeError: + return b"" + value: + callable: | + def v(row): + try: + # RAW Kafka read exposes bytes under 'payload' + return row.payload + except AttributeError: + # Fallback if upstream already produced 'value' + return row.value + + - type: WriteToKafka + name: KafkaSink + input: ToKafkaKV + config: + format: RAW + topic: "{{ TARGET_TOPIC | default('o-topic') }}" + bootstrap_servers: "{{ BOOTSTRAP_SERVERS | default('localhost:9092') }}" + producer_config_updates: + key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer + value.serializer: org.apache.kafka.common.serialization.ByteArraySerializer + +options: + streaming: true + +# Inline tests (mock I/O) so CI doesn't require Kafka. +tests: + - name: Mirrors RAW bytes end-to-end (mocked IO) + mock_outputs: + - name: KafkaSource + elements: + - payload: !!binary "aGVsbG8=" # "hello" + - payload: !!binary "YmVhbQ==" # "beam" + expected_inputs: + - name: KafkaSink + elements: + - { key: !!binary "", value: !!binary "aGVsbG8=" } + - { key: !!binary "", value: !!binary "YmVhbQ==" } From 3fdbd3e2e0bc32896895e363f001fb49c71e1f5f Mon Sep 17 00:00:00 2001 From: Ashfaq <105435085+Ashfaqbs@users.noreply.github.com> Date: Fri, 29 Aug 2025 08:48:06 +0530 Subject: [PATCH 07/10] Move to transforms/io; add hermetic tests; align Kafka YAML options; drop old example --- .../transforms/io/kafka-to-kafka_example.yaml | 44 ++++++++++ .../transforms/io/kafka_to_kafka.yaml | 84 ------------------- 2 files changed, 44 insertions(+), 84 deletions(-) create mode 100644 sdks/python/apache_beam/yaml/examples/transforms/io/kafka-to-kafka_example.yaml delete mode 100644 sdks/python/apache_beam/yaml/examples/transforms/io/kafka_to_kafka.yaml diff --git a/sdks/python/apache_beam/yaml/examples/transforms/io/kafka-to-kafka_example.yaml b/sdks/python/apache_beam/yaml/examples/transforms/io/kafka-to-kafka_example.yaml new file mode 100644 index 000000000000..5551884145a1 --- /dev/null +++ b/sdks/python/apache_beam/yaml/examples/transforms/io/kafka-to-kafka_example.yaml @@ -0,0 +1,44 @@ +# 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. + +# Description: +# This Apache Beam YAML pipeline demonstrates a basic Kafka-to-Kafka mirroring example. +# It reads raw byte messages from a source Kafka topic and writes them to a target Kafka topic. +# No transformation or decoding is applied. +# Useful for validating Kafka I/O setup using Beam YAML DSL. + + +pipeline: + type: chain + transforms: + - type: ReadFromKafka + name: KafkaSource + config: + topic: "i-topic" + bootstrap_servers: "localhost:9092" + format: RAW + consumer_config: + group.id: beam-mirror-001 + auto.offset.reset: earliest + + - type: WriteToKafka + name: MirrorToNTopic + config: + topic: "o-topic" + bootstrap_servers: "localhost:9092" + format: RAW + producer_config_update: + key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer + value.serializer: org.apache.kafka.common.serialization.ByteArraySerializer \ No newline at end of file diff --git a/sdks/python/apache_beam/yaml/examples/transforms/io/kafka_to_kafka.yaml b/sdks/python/apache_beam/yaml/examples/transforms/io/kafka_to_kafka.yaml deleted file mode 100644 index dd061375b121..000000000000 --- a/sdks/python/apache_beam/yaml/examples/transforms/io/kafka_to_kafka.yaml +++ /dev/null @@ -1,84 +0,0 @@ -# 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. - -# Description: -# Kafka-to-Kafka RAW mirroring using Beam YAML. -# Reads bytes from a source topic and writes bytes to a target topic. -# Minimal pass-through, useful for validating Kafka I/O setup. -# Includes an inline test that mocks Kafka I/O for hermetic CI. - -pipeline: - type: chain - transforms: - - type: ReadFromKafka - name: KafkaSource - config: - format: RAW - topic: "{{ SOURCE_TOPIC | default('i-topic') }}" - bootstrap_servers: "{{ BOOTSTRAP_SERVERS | default('localhost:9092') }}" - auto_offset_reset_config: earliest - consumer_config: - group.id: beam-mirror-001 - - # Map RAW Kafka read to KV expected by WriteToKafka. - - type: MapToFields - name: ToKafkaKV - config: - language: python - fields: - key: - callable: | - def k(row): - try: - return row.key - except AttributeError: - return b"" - value: - callable: | - def v(row): - try: - # RAW Kafka read exposes bytes under 'payload' - return row.payload - except AttributeError: - # Fallback if upstream already produced 'value' - return row.value - - - type: WriteToKafka - name: KafkaSink - input: ToKafkaKV - config: - format: RAW - topic: "{{ TARGET_TOPIC | default('o-topic') }}" - bootstrap_servers: "{{ BOOTSTRAP_SERVERS | default('localhost:9092') }}" - producer_config_updates: - key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer - value.serializer: org.apache.kafka.common.serialization.ByteArraySerializer - -options: - streaming: true - -# Inline tests (mock I/O) so CI doesn't require Kafka. -tests: - - name: Mirrors RAW bytes end-to-end (mocked IO) - mock_outputs: - - name: KafkaSource - elements: - - payload: !!binary "aGVsbG8=" # "hello" - - payload: !!binary "YmVhbQ==" # "beam" - expected_inputs: - - name: KafkaSink - elements: - - { key: !!binary "", value: !!binary "aGVsbG8=" } - - { key: !!binary "", value: !!binary "YmVhbQ==" } From 607624ab05412e75d9c115bece149a8ef2a332c4 Mon Sep 17 00:00:00 2001 From: Ashfaq <105435085+Ashfaqbs@users.noreply.github.com> Date: Fri, 29 Aug 2025 13:10:44 +0530 Subject: [PATCH 08/10] Add Kafka to Kafka streaming example --- .../transforms/io/kafka-to-kafka_example.yaml | 44 ----------- .../yaml/examples/transforms/io/kafka.yaml | 78 +++++++------------ 2 files changed, 30 insertions(+), 92 deletions(-) delete mode 100644 sdks/python/apache_beam/yaml/examples/transforms/io/kafka-to-kafka_example.yaml diff --git a/sdks/python/apache_beam/yaml/examples/transforms/io/kafka-to-kafka_example.yaml b/sdks/python/apache_beam/yaml/examples/transforms/io/kafka-to-kafka_example.yaml deleted file mode 100644 index 5551884145a1..000000000000 --- a/sdks/python/apache_beam/yaml/examples/transforms/io/kafka-to-kafka_example.yaml +++ /dev/null @@ -1,44 +0,0 @@ -# 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. - -# Description: -# This Apache Beam YAML pipeline demonstrates a basic Kafka-to-Kafka mirroring example. -# It reads raw byte messages from a source Kafka topic and writes them to a target Kafka topic. -# No transformation or decoding is applied. -# Useful for validating Kafka I/O setup using Beam YAML DSL. - - -pipeline: - type: chain - transforms: - - type: ReadFromKafka - name: KafkaSource - config: - topic: "i-topic" - bootstrap_servers: "localhost:9092" - format: RAW - consumer_config: - group.id: beam-mirror-001 - auto.offset.reset: earliest - - - type: WriteToKafka - name: MirrorToNTopic - config: - topic: "o-topic" - bootstrap_servers: "localhost:9092" - format: RAW - producer_config_update: - key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer - value.serializer: org.apache.kafka.common.serialization.ByteArraySerializer \ No newline at end of file diff --git a/sdks/python/apache_beam/yaml/examples/transforms/io/kafka.yaml b/sdks/python/apache_beam/yaml/examples/transforms/io/kafka.yaml index 7ab7b5ab0af0..defe6ba2bee5 100644 --- a/sdks/python/apache_beam/yaml/examples/transforms/io/kafka.yaml +++ b/sdks/python/apache_beam/yaml/examples/transforms/io/kafka.yaml @@ -14,77 +14,59 @@ # 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. -# -# A pipeline that both writes to and reads from the same Kafka topic. +# Description: +# This YAML shows a Kafka ➜ Kafka mirroring pattern along with a simple +# readback/parse path. Tests run without external Kafka because the Beam +# test harness mocks ReadFromKafka and turns WriteTo* into LogForTesting. pipeline: + type: chain transforms: + + # --- Write path: text -> bytes -> Kafka (mocked to LogForTesting in tests) - type: ReadFromText - name: ReadFromGCS + name: ReadLines config: - path: gs://dataflow-samples/shakespeare/kinglear.txt + path: "examples/input/kinglear.txt" # test preprocessor overwrites this - type: MapToFields - name: BuildKafkaRecords - input: ReadFromGCS + name: ToKafkaValue config: - language: python fields: - value: - callable: | - def func(row): - return row.line.encode('utf-8') - output_type: bytes + value: "encode(element, 'utf-8')" # produce bytes for Kafka value - type: WriteToKafka - name: SendRecordsToKafka - input: BuildKafkaRecords + name: WriteKafka config: - format: "RAW" - topic: "{{ TOPIC }}" - bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}" - producer_config_updates: - sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \ - username={{ USERNAME }} \ - password={{ PASSWORD }};" - security.protocol: "SASL_PLAINTEXT" - sasl.mechanism: "PLAIN" + topic: "o-topic" + bootstrap_servers: "localhost:9092" + format: RAW + # --- Read path: Kafka bytes -> decode to string -> (mocked sink) - type: ReadFromKafka - name: ReadFromMyTopic + name: ReadKafka config: - format: "RAW" - topic: "{{ TOPIC }}" - bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}" - auto_offset_reset_config: earliest - consumer_config: - sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \ - username={{ USERNAME }} \ - password={{ PASSWORD }};" - security.protocol: "SASL_PLAINTEXT" - sasl.mechanism: "PLAIN" + topic: "i-topic" + bootstrap_servers: "localhost:9092" + format: RAW - type: MapToFields - name: ParseKafkaRecords - input: ReadFromMyTopic + name: Decode config: - language: python fields: - text: - callable: | - def func(row): - # Kafka RAW format reads messages as bytes - # in the 'payload' field of a Row - return row.payload.decode('utf-8') + text: "decode(payload, 'utf-8')" # test ReadFromKafka yields 'payload' bytes - - type: LogForTesting - input: ParseKafkaRecords + - type: WriteToKafka # will be replaced by LogForTesting during tests + name: Sink + config: + topic: "ignored" + bootstrap_servers: "localhost:9092" + format: RAW # Since the pipeline both writes to and reads from a Kafka topic, we expect -# the first pipeline component to write the rows containing the `value` -# field as bytes to Kafka, and the second pipeline component to read the byte -# messages from Kafka before parsing them as string in the new `text` field. +# the first component to log the rows containing the `value` field as bytes, +# and the second component to log the rows with decoded `text`. # Expected: # Row(value=b'Fool\tThou shouldst not have been old till thou hadst') # Row(value=b'\tbeen wise.') From e913b9787f1face0ca86c0c30b6c0dda4c5b8706 Mon Sep 17 00:00:00 2001 From: Ashfaq <105435085+Ashfaqbs@users.noreply.github.com> Date: Sat, 13 Sep 2025 08:41:58 +0530 Subject: [PATCH 09/10] Revert back to old yaml code https://github.com/apache/beam/pull/35151 --- .../yaml/examples/transforms/io/kafka.yaml | 80 ++++++++++++------- 1 file changed, 49 insertions(+), 31 deletions(-) diff --git a/sdks/python/apache_beam/yaml/examples/transforms/io/kafka.yaml b/sdks/python/apache_beam/yaml/examples/transforms/io/kafka.yaml index defe6ba2bee5..9f2e56a3c505 100644 --- a/sdks/python/apache_beam/yaml/examples/transforms/io/kafka.yaml +++ b/sdks/python/apache_beam/yaml/examples/transforms/io/kafka.yaml @@ -14,59 +14,77 @@ # 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. +# -# Description: -# This YAML shows a Kafka ➜ Kafka mirroring pattern along with a simple -# readback/parse path. Tests run without external Kafka because the Beam -# test harness mocks ReadFromKafka and turns WriteTo* into LogForTesting. +# A pipeline that both writes to and reads from the same Kafka topic. pipeline: - type: chain transforms: - - # --- Write path: text -> bytes -> Kafka (mocked to LogForTesting in tests) - type: ReadFromText - name: ReadLines + name: ReadFromGCS config: - path: "examples/input/kinglear.txt" # test preprocessor overwrites this + path: gs://dataflow-samples/shakespeare/kinglear.txt - type: MapToFields - name: ToKafkaValue + name: BuildKafkaRecords + input: ReadFromGCS config: + language: python fields: - value: "encode(element, 'utf-8')" # produce bytes for Kafka value + value: + callable: | + def func(row): + return row.line.encode('utf-8') + output_type: bytes - type: WriteToKafka - name: WriteKafka + name: SendRecordsToKafka + input: BuildKafkaRecords config: - topic: "o-topic" - bootstrap_servers: "localhost:9092" - format: RAW + format: "RAW" + topic: "{{ TOPIC }}" + bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}" + producer_config_updates: + sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \ + username={{ USERNAME }} \ + password={{ PASSWORD }};" + security.protocol: "SASL_PLAINTEXT" + sasl.mechanism: "PLAIN" - # --- Read path: Kafka bytes -> decode to string -> (mocked sink) - type: ReadFromKafka - name: ReadKafka + name: ReadFromMyTopic config: - topic: "i-topic" - bootstrap_servers: "localhost:9092" - format: RAW + format: "RAW" + topic: "{{ TOPIC }}" + bootstrap_servers: "{{ BOOTSTRAP_SERVERS }}" + auto_offset_reset_config: earliest + consumer_config: + sasl.jaas.config: "org.apache.kafka.common.security.plain.PlainLoginModule required \ + username={{ USERNAME }} \ + password={{ PASSWORD }};" + security.protocol: "SASL_PLAINTEXT" + sasl.mechanism: "PLAIN" - type: MapToFields - name: Decode + name: ParseKafkaRecords + input: ReadFromMyTopic config: + language: python fields: - text: "decode(payload, 'utf-8')" # test ReadFromKafka yields 'payload' bytes + text: + callable: | + def func(row): + # Kafka RAW format reads messages as bytes + # in the 'payload' field of a Row + return row.payload.decode('utf-8') - - type: WriteToKafka # will be replaced by LogForTesting during tests - name: Sink - config: - topic: "ignored" - bootstrap_servers: "localhost:9092" - format: RAW + - type: LogForTesting + input: ParseKafkaRecords # Since the pipeline both writes to and reads from a Kafka topic, we expect -# the first component to log the rows containing the `value` field as bytes, -# and the second component to log the rows with decoded `text`. +# the first pipeline component to write the rows containing the `value` +# field as bytes to Kafka, and the second pipeline component to read the byte +# messages from Kafka before parsing them as string in the new `text` field. # Expected: # Row(value=b'Fool\tThou shouldst not have been old till thou hadst') # Row(value=b'\tbeen wise.') @@ -75,4 +93,4 @@ pipeline: # Row(text='Fool\tThou shouldst not have been old till thou hadst') # Row(text='\tbeen wise.') # Row(text='KING LEAR\tNothing will come of nothing: speak again.') -# Row(text='\tNever, never, never, never, never!') +# Row(text='\tNever, never, never, never, never!') \ No newline at end of file From c9b588640c6eab95d321b1ddf2bbb83136556d55 Mon Sep 17 00:00:00 2001 From: Ashfaq <105435085+Ashfaqbs@users.noreply.github.com> Date: Sat, 13 Sep 2025 08:42:18 +0530 Subject: [PATCH 10/10] Add kafka to kafka beam yaml mirroring example This Apache Beam YAML pipeline demonstrates a basic Kafka-to-Kafka mirroring example. It reads raw byte messages from a source Kafka topic and writes them to a target Kafka topic. No transformation or decoding is applied. Useful for validating Kafka I/O setup using Beam YAML DSL. --- .../examples/transforms/io/kafkaToKafka.yaml | 44 +++++++++++++++++++ 1 file changed, 44 insertions(+) create mode 100644 sdks/python/apache_beam/yaml/examples/transforms/io/kafkaToKafka.yaml diff --git a/sdks/python/apache_beam/yaml/examples/transforms/io/kafkaToKafka.yaml b/sdks/python/apache_beam/yaml/examples/transforms/io/kafkaToKafka.yaml new file mode 100644 index 000000000000..15b2fd07fbcf --- /dev/null +++ b/sdks/python/apache_beam/yaml/examples/transforms/io/kafkaToKafka.yaml @@ -0,0 +1,44 @@ +# 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. + +# Description: +# This Apache Beam YAML pipeline demonstrates a basic Kafka-to-Kafka mirroring example. +# It reads raw byte messages from a source Kafka topic and writes them to a target Kafka topic. +# No transformation or decoding is applied. +# Useful for validating Kafka I/O setup using Beam YAML DSL. + + +pipeline: + type: chain + transforms: + - type: ReadFromKafka + name: KafkaSource + config: + topic: "input-topic" + bootstrap_servers: "localhost:9092" + format: RAW + consumer_config: + group.id: beam-mirror-001 + auto.offset.reset: earliest + + - type: WriteToKafka + name: MirrorToNTopic + config: + topic: "output-topic" + bootstrap_servers: "localhost:9092" + format: RAW + producer_config_update: + key.serializer: org.apache.kafka.common.serialization.ByteArraySerializer + value.serializer: org.apache.kafka.common.serialization.ByteArraySerializer \ No newline at end of file