[KafkaIO] Improve caching in backlog estimation and processing (#34331)

* Improve caching in backlog estimation and processing

* Add comment to explain the behavior of volatile guard field in KafkaLatestOffsetEstimator

* Guard against exceptions in endOffset refresh

* Call cancelIfTimeouted in roundtripElements to shutdown lingering pipelines

* Add missing calls to seek and/or pause before return points added in #34202

Author: sjvanrossum

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

@@ -18,6 +18,7 @@
 package org.apache.beam.sdk.io.kafka;
 
 import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
+import static org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
 
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
@@ -130,6 +131,20 @@ static Map<String, Object> getOffsetConsumerConfig(
     return offsetConsumerConfig;
   }
 
+  static Map<String, Object> overrideBootstrapServersConfig(
+      Map<String, Object> currentConfig, KafkaSourceDescriptor description) {
+    checkState(
+        currentConfig.containsKey(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG)
+            || description.getBootStrapServers() != null);
+    Map<String, Object> config = new HashMap<>(currentConfig);
+    if (description.getBootStrapServers() != null && description.getBootStrapServers().size() > 0) {
+      config.put(
+          ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
+          String.join(",", description.getBootStrapServers()));
+    }
+    return config;
+  }
+
   /*
    * Maintains approximate average over last 1000 elements.
    * Usage is only thread-safe for a single producer and multiple consumers.