Address review comments

Author: bbejeck

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

@@ -500,7 +500,7 @@ public static StreamThread create(final TopologyMetadata topologyMetadata,
             stateDirectory,
             stateUpdater,
             schedulingTaskManager,
-            maxPollIntervalMs / 2
+            maxPollIntervalMs * 3 / 4
         );
         referenceContainer.taskManager = taskManager;
 

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

@@ -111,6 +111,7 @@ public class TaskManager {
     private final StateUpdater stateUpdater;
     private final DefaultTaskManager schedulingTaskManager;
     private final long waitForFutureTimeoutMs;
+    
     TaskManager(final Time time,
                 final ChangelogReader changelogReader,
                 final ProcessId processId,

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

@@ -4207,14 +4207,14 @@ public void shouldSetWaitForFutureTimeoutFromMaxPollIntervalMs() {
         final StreamsConfig config = new StreamsConfig(properties);
         thread = createStreamThread(CLIENT_ID, config);
 
-        assertThat(thread.taskManager().waitForFutureTimeoutMs(), equalTo(10_000L));
+        assertThat(thread.taskManager().waitForFutureTimeoutMs(), equalTo(15_000L));
     }
 
     @Test
     public void shouldSetDefaultWaitForFutureTimeoutFromDefaultMaxPollIntervalMs() {
         thread = createStreamThread(CLIENT_ID, false);
 
-        assertThat(thread.taskManager().waitForFutureTimeoutMs(), equalTo(150_000L));
+        assertThat(thread.taskManager().waitForFutureTimeoutMs(), equalTo(225_000L));
     }
 
     private boolean runUntilTimeoutOrCondition(final Runnable action, final TestCondition testCondition) throws Exception {