KAFKA-12738: address minor followup and consolidate integration tests of PR #11787

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

@@ -548,6 +548,7 @@ public void run() {
             failedStreamThreadSensor.record();
             requestLeaveGroupDuringShutdown();
             streamsUncaughtExceptionHandler.accept(e, false);
+            // Note: the above call currently rethrows the exception, so nothing below this line will be executed
         } finally {
             completeShutdown(cleanRun);
         }

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

@@ -31,6 +31,9 @@
  * shared between all StreamThreads.
  */
 public class TaskExecutionMetadata {
+    // TODO: implement exponential backoff, for now we just wait 5s
+    private static final long CONSTANT_BACKOFF_MS = 5_000L;
+
     private final boolean hasNamedTopologies;
     // map of topologies experiencing errors/currently under backoff
     private final ConcurrentHashMap<String, NamedTopologyMetadata> topologyNameToErrorMetadata = new ConcurrentHashMap<>();
@@ -58,7 +61,7 @@ public void registerTaskError(final Task task, final Throwable t, final long now
         }
     }
 
-    class NamedTopologyMetadata {
+    private class NamedTopologyMetadata {
         private final Logger log;
         private final Map<TaskId, Long> tasksToErrorTime = new ConcurrentHashMap<>();
 
@@ -73,11 +76,10 @@ public boolean canProcess() {
         }
 
         public boolean canProcessTask(final Task task, final long now) {
-            // TODO: implement exponential backoff, for now we just wait 15s
             final Long errorTime = tasksToErrorTime.get(task.id());
             if (errorTime == null) {
                 return true;
-            } else if (now - errorTime > 15000L) {
+            } else if (now - errorTime > CONSTANT_BACKOFF_MS) {
                 log.info("End backoff for task {} at t={}", task.id(), now);
                 tasksToErrorTime.remove(task.id());
                 if (tasksToErrorTime.isEmpty()) {