Added test confirming correct math

Author: bbejeck

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

@@ -483,10 +483,9 @@ public static StreamThread create(final TopologyMetadata topologyMetadata,
                 threadIdx
             );
 
-        final int dummyThreadIdxForConfig = 1;
-        final long maxPollIntervalMs = new InternalConsumerConfig(
-            config.getMainConsumerConfigs("dummyGroupId", "dummyClientId", dummyThreadIdxForConfig))
-            .getInt(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG);
+        final long maxPollIntervalMs = Integer.parseInt(
+            config.getMainConsumerConfigs("dummy", "dummy", threadIdx)
+                .getOrDefault(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "300000").toString());
 
         final TaskManager taskManager = new TaskManager(
             time,

streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java

@@ -158,6 +158,11 @@ void setMainConsumer(final Consumer<byte[], byte[]> mainConsumer) {
         this.mainConsumer = mainConsumer;
     }
 
+    // visible for testing
+    long waitForFutureTimeoutMs() {
+        return waitForFutureTimeoutMs;
+    }
+
     public double totalProducerBlockedTime() {
         return activeTaskCreator.totalProducerBlockedTime();
     }

streams/src/test/java/org/apache/kafka/streams/processor/internals/StreamThreadTest.java

@@ -18,6 +18,7 @@
 
 import org.apache.kafka.clients.admin.MockAdminClient;
 import org.apache.kafka.clients.consumer.Consumer;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -4199,6 +4200,23 @@ private void runUntilTimeoutOrException(final Runnable action) {
         }
     }
 
+    @Test
+    public void shouldSetWaitForFutureTimeoutFromMaxPollIntervalMs() {
+        final Properties properties = configProps(false, false);
+        properties.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, "20000");
+        final StreamsConfig config = new StreamsConfig(properties);
+        thread = createStreamThread(CLIENT_ID, config);
+
+        assertThat(thread.taskManager().waitForFutureTimeoutMs(), equalTo(10_000L));
+    }
+
+    @Test
+    public void shouldSetDefaultWaitForFutureTimeoutFromDefaultMaxPollIntervalMs() {
+        thread = createStreamThread(CLIENT_ID, false);
+
+        assertThat(thread.taskManager().waitForFutureTimeoutMs(), equalTo(150_000L));
+    }
+
     private boolean runUntilTimeoutOrCondition(final Runnable action, final TestCondition testCondition) throws Exception {
         final long expectedEnd = System.currentTimeMillis() + DEFAULT_MAX_WAIT_MS;
         while (System.currentTimeMillis() < expectedEnd) {