Fix cases where reader was not overwritten when building Kafka reader.

tomstepp · tomstepp · commit da5ed2672e4b · 2025-09-22T21:18:17.000-07:00
diff --git a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
@@ -492,10 +492,10 @@ static KafkaIO.Read<Integer, Long> mkKafkaReadTransformBase(
         reader = reader.withRedistributeNumKeys(numKeys);
       }
       if (offsetDeduplication != null && offsetDeduplication) {
-        reader.withOffsetDeduplication(offsetDeduplication);
+        reader = reader.withOffsetDeduplication(offsetDeduplication);
       }
       if (redistributeByRecordKey != null && redistributeByRecordKey) {
-        reader.withRedistributeByRecordKey(redistributeByRecordKey);
+        reader = reader.withRedistributeByRecordKey(redistributeByRecordKey);
       }
     }
     return reader;

Original file line number	Diff line number	Diff line change
`@@ -492,10 +492,10 @@ static KafkaIO.Read<Integer, Long> mkKafkaReadTransformBase(`
`492`	`492`	`reader = reader.withRedistributeNumKeys(numKeys);`
`493`	`493`	`}`
`494`	`494`	`if (offsetDeduplication != null && offsetDeduplication) {`
`495`		`- reader.withOffsetDeduplication(offsetDeduplication);`
	`495`	`+ reader = reader.withOffsetDeduplication(offsetDeduplication);`
`496`	`496`	`}`
`497`	`497`	`if (redistributeByRecordKey != null && redistributeByRecordKey) {`
`498`		`- reader.withRedistributeByRecordKey(redistributeByRecordKey);`
	`498`	`+ reader = reader.withRedistributeByRecordKey(redistributeByRecordKey);`
`499`	`499`	`}`
`500`	`500`	`}`
`501`	`501`	`return reader;`