(beam) branch release-2.58.1 updated: Cherry pick (#32202)

[da...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

damccorm pushed a commit to branch release-2.58.1
in repository https://gitbox.apache.org/repos/asf/beam.git


The following commit(s) were added to refs/heads/release-2.58.1 by this push:
     new 57e5a46300c Cherry pick (#32202)
57e5a46300c is described below

commit 57e5a46300c8833d4604291bb92003bc656e4186
Author: Danny McCormick <da...@google.com>
AuthorDate: Thu Aug 15 19:58:13 2024 +0200

    Cherry pick (#32202)
    
    Co-authored-by: scwhittle <sc...@users.noreply.github.com>
---
 CHANGES.md                                                            | 4 +++-
 .../io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java  | 2 --
 2 files changed, 3 insertions(+), 3 deletions(-)

diff --git a/CHANGES.md b/CHANGES.md
index 243596e6f2e..08442ce09bf 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -92,7 +92,9 @@
 
 ## Known Issues
 
-* ([#X](https://github.com/apache/beam/issues/X)).
+* Large Dataflow graphs using runner v2, or pipelines explicitly enabling the `upload_graph` experiment, will fail at construction time ([#32159](https://github.com/apache/beam/issues/32159)).
+* Python pipelines that run with 2.53.0-2.58.0 SDKs and read data from GCS might be affected by a data corruption issue ([#32169](https://github.com/apache/beam/issues/32169)). The issue will be fixed in 2.59.0 ([#32135](https://github.com/apache/beam/pull/32135)). To work around this, update the google-cloud-storage package to version 2.18.2 or newer.
+* [KafkaIO] Records read with `ReadFromKafkaViaSDF` are redistributed and may contain duplicates regardless of the configuration. This affects Java pipelines with Dataflow v2 runner and xlang pipelines reading from Kafka, ([#32196](https://github.com/apache/beam/issues/32196))
 
 # [2.57.0] - 2024-06-26
 
diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
index 68400e83106..c1526d5382b 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
@@ -1730,8 +1730,6 @@ public class KafkaIO {
                 .withKeyDeserializerProvider(kafkaRead.getKeyDeserializerProvider())
                 .withValueDeserializerProvider(kafkaRead.getValueDeserializerProvider())
                 .withManualWatermarkEstimator()
-                .withRedistribute()
-                .withAllowDuplicates() // must be set with withRedistribute option.
                 .withTimestampPolicyFactory(kafkaRead.getTimestampPolicyFactory())
                 .withCheckStopReadingFn(kafkaRead.getCheckStopReadingFn())
                 .withConsumerPollingTimeout(kafkaRead.getConsumerPollingTimeout());