Skip to content

Commit 5b14dfd

Browse files
committed
Revert "Apply offset consumer config overrides to admin client config"
This reverts commit 5f4bcbe.
1 parent fd9ca40 commit 5b14dfd

1 file changed

Lines changed: 1 addition & 11 deletions

File tree

sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java

Lines changed: 1 addition & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -24,7 +24,6 @@
2424
import java.math.MathContext;
2525
import java.time.Duration;
2626
import java.util.Collections;
27-
import java.util.HashMap;
2827
import java.util.List;
2928
import java.util.Map;
3029
import java.util.Optional;
@@ -211,15 +210,6 @@ private ReadFromKafkaDoFn(
211210
transform.getAdminFactoryFn();
212211
final SerializableFunction<Map<String, Object>, Consumer<byte[], byte[]>> consumerFactoryFn =
213212
transform.getConsumerFactoryFn();
214-
final @Nullable Map<String, Object> offsetConsumerConfigOverrides =
215-
transform.getOffsetConsumerConfig();
216-
final Map<String, Object> offsetConsumerConfig;
217-
if (offsetConsumerConfigOverrides == null) {
218-
offsetConsumerConfig = transform.getConsumerConfig();
219-
} else {
220-
offsetConsumerConfig = new HashMap<>(transform.getConsumerConfig());
221-
offsetConsumerConfig.putAll(offsetConsumerConfigOverrides);
222-
}
223213
this.consumerConfig = transform.getConsumerConfig();
224214
this.keyDeserializerProvider =
225215
Preconditions.checkArgumentNotNull(transform.getKeyDeserializerProvider());
@@ -270,7 +260,7 @@ public KafkaLatestOffsetEstimator load(
270260
sourceDescriptor);
271261
final Map<String, Object> config =
272262
KafkaIOUtils.overrideBootstrapServersConfig(
273-
offsetConsumerConfig, sourceDescriptor);
263+
consumerConfig, sourceDescriptor);
274264
final Admin admin = adminFactoryFn.apply(config);
275265
return new KafkaLatestOffsetEstimator(
276266
admin, sourceDescriptor.getTopicPartition());

0 commit comments

Comments
 (0)