Apply offset consumer config overrides to admin client config

Author: sjvanrossum

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

@@ -24,6 +24,7 @@
 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;
@@ -200,6 +201,15 @@ 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());
@@ -250,7 +260,7 @@ public KafkaLatestOffsetEstimator load(
                                 sourceDescriptor);
                             final Map<String, Object> config =
                                 KafkaIOUtils.overrideBootstrapServersConfig(
-                                    consumerConfig, sourceDescriptor);
+                                    offsetConsumerConfig, sourceDescriptor);
                             final Admin admin = adminFactoryFn.apply(config);
                             return new KafkaLatestOffsetEstimator(
                                 admin, sourceDescriptor.getTopicPartition());