Revert "Apply offset consumer config overrides to admin client config"

This reverts commit 5f4bcbe.

Author: sjvanrossum

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

@@ -24,7 +24,6 @@
 import java.math.MathContext;
 import java.time.Duration;
 import java.util.Collections;
-import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -201,15 +200,6 @@ private ReadFromKafkaDoFn(
         transform.getAdminFactoryFn();
     final SerializableFunction<Map<String, Object>, Consumer<byte[], byte[]>> consumerFactoryFn =
         transform.getConsumerFactoryFn();
-    final @Nullable Map<String, Object> offsetConsumerConfigOverrides =
-        transform.getOffsetConsumerConfig();
-    final Map<String, Object> offsetConsumerConfig;
-    if (offsetConsumerConfigOverrides == null) {
-      offsetConsumerConfig = transform.getConsumerConfig();
-    } else {
-      offsetConsumerConfig = new HashMap<>(transform.getConsumerConfig());
-      offsetConsumerConfig.putAll(offsetConsumerConfigOverrides);
-    }
     this.consumerConfig = transform.getConsumerConfig();
     this.keyDeserializerProvider =
         Preconditions.checkArgumentNotNull(transform.getKeyDeserializerProvider());
@@ -260,7 +250,7 @@ public KafkaLatestOffsetEstimator load(
                                 sourceDescriptor);
                             final Map<String, Object> config =
                                 KafkaIOUtils.overrideBootstrapServersConfig(
-                                    offsetConsumerConfig, sourceDescriptor);
+                                    consumerConfig, sourceDescriptor);
                             final Admin admin = adminFactoryFn.apply(config);
                             return new KafkaLatestOffsetEstimator(
                                 admin, sourceDescriptor.getTopicPartition());