Address review: explicit fallback gates, single-deadline allOf

- Replace throw-and-catch in resolver with explicit if-gate: a
  non-positive replica count (zero, negative, or absent) logs and
  returns FALLBACK_NUM_STREAM_THREADS. optionalReplicaCount() now
  yields 0 instead of throwing.
- Calculator: replace per-topic get(timeout) loop with
  KafkaFuture.allOf(...).get(timeout) so worst-case startup wait is
  one deadline instead of N. Per-future UnknownTopicOrPartitionException
  handling is preserved.
- KafkaStreamsApp: when num.stream.threads = DYNAMIC and no resolver
  is provided (or the override throws), substitute
  FALLBACK_NUM_STREAM_THREADS so the literal "DYNAMIC" never reaches
  Kafka Streams config (it can't parse the string).
- Promote FALLBACK_NUM_STREAM_THREADS to public so callers can
  reference the same constant.
- Document regex/pattern subscription limitation in the calculator
  javadoc.
- Document fallback rationale on FALLBACK_NUM_STREAM_THREADS.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: suresh-prakash

kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/KafkaStreamsApp.java

@@ -36,7 +36,6 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.OptionalInt;
 import java.util.Properties;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
@@ -110,8 +109,9 @@ protected void doInit() {
       streamsBuilder = buildTopology(streamsConfig, streamsBuilder, sourceStreams);
       this.topology = streamsBuilder.build();
 
-      resolveDynamicStreamThreads(streamsConfig)
-          .ifPresent(threads -> streamsConfig.put(NUM_STREAM_THREADS_CONFIG, threads));
+      if (StreamThreadsCountResolver.isDynamic(streamsConfig)) {
+        streamsConfig.put(NUM_STREAM_THREADS_CONFIG, resolveDynamicStreamThreads(streamsConfig));
+      }
 
       getLogger().info("Finalized kafka streams configuration: {}", streamsConfig);
 
@@ -292,31 +292,31 @@ protected Optional<StreamThreadsCountResolver> getStreamThreadsCountResolver() {
     return Optional.empty();
   }
 
-  // Defensive: any unexpected runtime failure in dynamic resolution must NOT prevent the app from
-  // starting. If something goes wrong, log and return empty so the caller leaves the streams
-  // config untouched. JVM Errors (OOM, LinkageError, etc.) propagate — those are not recoverable.
-  private OptionalInt resolveDynamicStreamThreads(Map<String, Object> streamsProperties) {
+  // Caller must check StreamThreadsCountResolver.isDynamic(...) before invoking. Always returns
+  // a concrete value: either the resolver's computed count or FALLBACK_NUM_STREAM_THREADS.
+  // The string sentinel "DYNAMIC" must never leak back to Kafka Streams config — it can't be
+  // parsed as an integer.
+  private int resolveDynamicStreamThreads(Map<String, Object> streamsProperties) {
+    Optional<StreamThreadsCountResolver> resolver;
     try {
-      if (!StreamThreadsCountResolver.isDynamic(streamsProperties)) {
-        return OptionalInt.empty();
-      }
-      Optional<StreamThreadsCountResolver> resolver = getStreamThreadsCountResolver();
-      if (resolver.isEmpty()) {
-        getLogger()
-            .warn(
-                "{} is set to DYNAMIC but no StreamThreadsCountResolver is provided; leaving as-is",
-                NUM_STREAM_THREADS_CONFIG);
-        return OptionalInt.empty();
-      }
-      return OptionalInt.of(resolver.get().resolve(this.topology, streamsProperties));
+      resolver = getStreamThreadsCountResolver();
     } catch (Exception exception) {
       getLogger()
-          .error(
-              "Unexpected error resolving dynamic {}; leaving config untouched",
-              NUM_STREAM_THREADS_CONFIG,
+          .warn(
+              "getStreamThreadsCountResolver() threw; falling back to {} stream threads",
+              StreamThreadsCountResolver.FALLBACK_NUM_STREAM_THREADS,
               exception);
-      return OptionalInt.empty();
+      return StreamThreadsCountResolver.FALLBACK_NUM_STREAM_THREADS;
+    }
+    if (resolver.isEmpty()) {
+      getLogger()
+          .warn(
+              "{} is set to DYNAMIC but no StreamThreadsCountResolver is provided; falling back to {}",
+              NUM_STREAM_THREADS_CONFIG,
+              StreamThreadsCountResolver.FALLBACK_NUM_STREAM_THREADS);
+      return StreamThreadsCountResolver.FALLBACK_NUM_STREAM_THREADS;
     }
+    return resolver.get().resolve(this.topology, streamsProperties);
   }
 
   public Map<String, Object> getStreamsConfig(Config jobConfig) {

kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/threading/DynamicStreamThreadsCountCalculator.java

@@ -29,6 +29,9 @@
  * <p>For each sub-topology the maximum partition count across its source topics is the number of
  * stream tasks. Summing across sub-topologies and dividing by the replica count yields the threads
  * each instance should run to keep all tasks active without idle threads.
+ *
+ * <p>Regex/pattern subscriptions ({@link Source#topicPattern()}) are not supported — those
+ * sub-topologies contribute zero tasks. No current app subscribes via regex.
  */
 public class DynamicStreamThreadsCountCalculator {
 
@@ -93,13 +96,19 @@ private Map<String, Integer> describePartitions(
       return Map.of();
     }
     final DescribeTopicsResult result = adminClient.describeTopics(topics);
+    final Map<String, KafkaFuture<TopicDescription>> futures = result.topicNameValues();
+
+    // Single shared deadline across all topics — worst-case wait is one timeout, not N.
+    // AdminClient retries the underlying RPC internally per its own retry config.
+    awaitAll(futures.values(), DESCRIBE_TOPICS_TIMEOUT_MILLIS);
+
     final Map<String, Integer> partitions = new HashMap<>();
-    for (final Entry<String, KafkaFuture<TopicDescription>> entry :
-        result.topicNameValues().entrySet()) {
+    for (final Entry<String, KafkaFuture<TopicDescription>> entry : futures.entrySet()) {
       try {
-        final TopicDescription description =
-            entry.getValue().get(DESCRIBE_TOPICS_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS);
-        partitions.put(entry.getKey(), description.partitions().size());
+        // Non-blocking after awaitAll: each future is already complete (or completed
+        // exceptionally). Per-topic handling stays so UnknownTopicOrPartitionException
+        // doesn't fail the whole batch.
+        partitions.put(entry.getKey(), entry.getValue().get().partitions().size());
       } catch (final ExecutionException executionException) {
         if (executionException.getCause() instanceof UnknownTopicOrPartitionException) {
           logger.warn(
@@ -114,11 +123,25 @@ private Map<String, Integer> describePartitions(
         Thread.currentThread().interrupt();
         throw new RuntimeException(
             "Interrupted while describing topic " + entry.getKey(), interruptedException);
-      } catch (final TimeoutException timeoutException) {
-        throw new RuntimeException(
-            "Timed out describing topic " + entry.getKey(), timeoutException);
       }
     }
     return Map.copyOf(partitions);
   }
+
+  private static void awaitAll(
+      final java.util.Collection<KafkaFuture<TopicDescription>> futures, final long timeoutMillis) {
+    try {
+      KafkaFuture.allOf(futures.toArray(KafkaFuture[]::new))
+          .get(timeoutMillis, TimeUnit.MILLISECONDS);
+    } catch (final TimeoutException timeoutException) {
+      throw new RuntimeException(
+          "Timed out describing topics after " + timeoutMillis + "ms", timeoutException);
+    } catch (final InterruptedException interruptedException) {
+      Thread.currentThread().interrupt();
+      throw new RuntimeException("Interrupted describing topics", interruptedException);
+    } catch (final ExecutionException executionException) {
+      // At least one topic future completed exceptionally; per-future handling below classifies
+      // (UnknownTopicOrPartitionException → 0 partitions; everything else → fail).
+    }
+  }
 }

kafka-streams-framework/src/main/java/org/hypertrace/core/kafkastreams/framework/threading/StreamThreadsCountResolver.java

@@ -17,12 +17,16 @@
  * resolution fails.
  *
  * <p>The replica count is supplied externally — applications typically wire it from a {@code
- * REPLICA_COUNT} environment variable injected by the deployment template.
+ * REPLICA_COUNT} environment variable injected by the deployment template. A non-positive value
+ * (zero, negative, or unset) falls back to {@link #FALLBACK_NUM_STREAM_THREADS}.
  */
 public class StreamThreadsCountResolver {
 
   public static final String DYNAMIC_SENTINEL = "DYNAMIC";
-  static final int FALLBACK_NUM_STREAM_THREADS = 8;
+
+  // Reasonable default for current Hypertrace Kafka Streams apps. Apps that need a different
+  // value should configure a numeric num.stream.threads explicitly instead of using DYNAMIC.
+  public static final int FALLBACK_NUM_STREAM_THREADS = 8;
 
   private static final Logger logger = LoggerFactory.getLogger(StreamThreadsCountResolver.class);
 
@@ -55,15 +59,21 @@ public static boolean isDynamic(final Map<String, Object> streamsProperties) {
 
   /**
    * Resolve a concrete thread count for the given topology. Returns {@link
-   * #FALLBACK_NUM_STREAM_THREADS} on any failure so the application can still start.
+   * #FALLBACK_NUM_STREAM_THREADS} when prerequisites are missing (non-positive replica count) or
+   * the AdminClient/calculator call fails, so the application can still start.
    */
   public int resolve(final Topology topology, final Map<String, Object> streamsProperties) {
-    try {
-      final int replicas = requirePositiveReplicaCount();
-      try (final AdminClient adminClient =
-          adminClientFactory.apply(toProperties(streamsProperties))) {
-        return calculator.compute(topology, adminClient, replicas);
-      }
+    final int replicas = replicaCountSupplier.getAsInt();
+    if (replicas <= 0) {
+      logger.warn(
+          "replica.count is non-positive ({}); falling back to {} stream threads",
+          replicas,
+          FALLBACK_NUM_STREAM_THREADS);
+      return FALLBACK_NUM_STREAM_THREADS;
+    }
+    try (final AdminClient adminClient =
+        adminClientFactory.apply(toProperties(streamsProperties))) {
+      return calculator.compute(topology, adminClient, replicas);
     } catch (final RuntimeException runtimeException) {
       logger.error(
           "Failed to compute dynamic num.stream.threads; falling back to {}",
@@ -75,12 +85,11 @@ public int resolve(final Topology topology, final Map<String, Object> streamsPro
 
   /**
    * Convenience adapter — given an {@code Optional<Integer>} replica-count source (the common
-   * config-loaded shape), produce a supplier that this resolver accepts.
+   * config-loaded shape), produce a supplier that yields the value when present and {@code 0} when
+   * absent (treated as a fallback signal by {@link #resolve}).
    */
   public static IntSupplier optionalReplicaCount(final Optional<Integer> replicaCount) {
-    return () ->
-        replicaCount.orElseThrow(
-            () -> new IllegalStateException("replica.count is not configured"));
+    return () -> replicaCount.orElse(0);
   }
 
   private static Properties toProperties(final Map<String, Object> streamsProperties) {
@@ -93,12 +102,4 @@ private static Properties toProperties(final Map<String, Object> streamsProperti
         });
     return properties;
   }
-
-  private int requirePositiveReplicaCount() {
-    final int replicas = replicaCountSupplier.getAsInt();
-    if (replicas <= 0) {
-      throw new IllegalStateException("replica.count must be positive, got " + replicas);
-    }
-    return replicas;
-  }
 }