Address review: inject AdminClient factory, narrow exception scope

- Invert resolver dep: take AdminClient factory in constructor instead
  of building inside resolve(). Removes the package-private test-only
  constructor and lets the test mock the actual external (AdminClient)
  while exercising the real DynamicStreamThreadsCountCalculator.
- Drop @mock on the calculator in the resolver test; route stubs
  through AdminClient.describeTopics(...) so the test asserts on the
  real partition→threads math instead of stubbed return values.
- Narrow catch (Throwable) to catch (Exception) in
  resolveDynamicStreamThreads so JVM Errors (OOM, LinkageError) keep
  surfacing instead of being silently swallowed at startup.
- Return OptionalInt from resolveDynamicStreamThreads and apply the
  override at the doInit() call site, so the helper no longer mutates
  the caller's map under a query-shaped name.

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,6 +36,7 @@
 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;
@@ -109,7 +110,8 @@ protected void doInit() {
       streamsBuilder = buildTopology(streamsConfig, streamsBuilder, sourceStreams);
       this.topology = streamsBuilder.build();
 
-      resolveDynamicStreamThreads(streamsConfig);
+      resolveDynamicStreamThreads(streamsConfig)
+          .ifPresent(threads -> streamsConfig.put(NUM_STREAM_THREADS_CONFIG, threads));
 
       getLogger().info("Finalized kafka streams configuration: {}", streamsConfig);
 
@@ -290,30 +292,30 @@ protected Optional<StreamThreadsCountResolver> getStreamThreadsCountResolver() {
     return Optional.empty();
   }
 
-  // Defensive: any unexpected failure in dynamic resolution must NOT prevent the app from starting.
-  // If something goes wrong, log and leave streamsProperties untouched so Kafka Streams uses
-  // whatever value was originally configured (or its own default).
-  private void resolveDynamicStreamThreads(Map<String, Object> streamsProperties) {
+  // 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) {
     try {
       if (!StreamThreadsCountResolver.isDynamic(streamsProperties)) {
-        return;
+        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;
+        return OptionalInt.empty();
       }
-      int resolved = resolver.get().resolve(this.topology, streamsProperties);
-      streamsProperties.put(NUM_STREAM_THREADS_CONFIG, resolved);
-    } catch (Throwable throwable) {
+      return OptionalInt.of(resolver.get().resolve(this.topology, streamsProperties));
+    } catch (Exception exception) {
       getLogger()
           .error(
               "Unexpected error resolving dynamic {}; leaving config untouched",
               NUM_STREAM_THREADS_CONFIG,
-              throwable);
+              exception);
+      return OptionalInt.empty();
     }
   }
 

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

@@ -3,6 +3,7 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.Properties;
+import java.util.function.Function;
 import java.util.function.IntSupplier;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.streams.Topology;
@@ -27,16 +28,19 @@ public class StreamThreadsCountResolver {
 
   private final DynamicStreamThreadsCountCalculator calculator;
   private final IntSupplier replicaCountSupplier;
+  private final Function<Properties, AdminClient> adminClientFactory;
 
   public StreamThreadsCountResolver(final IntSupplier replicaCountSupplier) {
-    this(new DynamicStreamThreadsCountCalculator(), replicaCountSupplier);
+    this(new DynamicStreamThreadsCountCalculator(), replicaCountSupplier, AdminClient::create);
   }
 
-  StreamThreadsCountResolver(
+  public StreamThreadsCountResolver(
       final DynamicStreamThreadsCountCalculator calculator,
-      final IntSupplier replicaCountSupplier) {
+      final IntSupplier replicaCountSupplier,
+      final Function<Properties, AdminClient> adminClientFactory) {
     this.calculator = calculator;
     this.replicaCountSupplier = replicaCountSupplier;
+    this.adminClientFactory = adminClientFactory;
   }
 
   /**
@@ -56,7 +60,8 @@ public static boolean isDynamic(final Map<String, Object> streamsProperties) {
   public int resolve(final Topology topology, final Map<String, Object> streamsProperties) {
     try {
       final int replicas = requirePositiveReplicaCount();
-      try (final AdminClient adminClient = AdminClient.create(toProperties(streamsProperties))) {
+      try (final AdminClient adminClient =
+          adminClientFactory.apply(toProperties(streamsProperties))) {
         return calculator.compute(topology, adminClient, replicas);
       }
     } catch (final RuntimeException runtimeException) {

kafka-streams-framework/src/test/java/org/hypertrace/core/kafkastreams/framework/threading/StreamThreadsCountResolverTest.java

@@ -1,45 +1,51 @@
 package org.hypertrace.core.kafkastreams.framework.threading;
 
+import static java.util.stream.Collectors.toUnmodifiableList;
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.mockito.ArgumentMatchers.any;
-import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.anySet;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Properties;
+import java.util.function.Function;
 import java.util.function.IntSupplier;
+import java.util.stream.IntStream;
 import org.apache.kafka.clients.CommonClientConfigs;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.DescribeTopicsResult;
+import org.apache.kafka.clients.admin.TopicDescription;
+import org.apache.kafka.common.KafkaFuture;
+import org.apache.kafka.common.Node;
+import org.apache.kafka.common.TopicPartitionInfo;
 import org.apache.kafka.common.serialization.Serdes;
 import org.apache.kafka.streams.StreamsBuilder;
 import org.apache.kafka.streams.StreamsConfig;
 import org.apache.kafka.streams.Topology;
 import org.apache.kafka.streams.kstream.Consumed;
 import org.junit.jupiter.api.Test;
-import org.junit.jupiter.api.extension.ExtendWith;
-import org.mockito.Mock;
-import org.mockito.junit.jupiter.MockitoExtension;
 
-@ExtendWith(MockitoExtension.class)
 class StreamThreadsCountResolverTest {
 
-  @Mock private DynamicStreamThreadsCountCalculator calculator;
-
   @Test
   void calculatorThrowFallsBackToEight() {
-    final Topology topology = simpleTopology();
-    when(calculator.compute(eq(topology), any(), eq(8)))
+    final AdminClient adminClient = mock(AdminClient.class);
+    when(adminClient.describeTopics(anySet()))
         .thenThrow(new RuntimeException("simulated broker outage"));
-    final StreamThreadsCountResolver resolver = new StreamThreadsCountResolver(calculator, () -> 8);
+    final StreamThreadsCountResolver resolver = resolverWith(() -> 8, adminClient);
 
-    final int resolved = resolver.resolve(topology, bootstrapProperties());
+    final int resolved = resolver.resolve(simpleTopology(), bootstrapProperties());
 
     assertEquals(StreamThreadsCountResolver.FALLBACK_NUM_STREAM_THREADS, resolved);
   }
 
   @Test
   void missingReplicaCountFallsBack() {
     final IntSupplier missing = StreamThreadsCountResolver.optionalReplicaCount(Optional.empty());
-    final StreamThreadsCountResolver resolver = new StreamThreadsCountResolver(calculator, missing);
+    final StreamThreadsCountResolver resolver = resolverWith(missing, mock(AdminClient.class));
 
     final int resolved = resolver.resolve(simpleTopology(), bootstrapProperties());
 
@@ -48,7 +54,7 @@ void missingReplicaCountFallsBack() {
 
   @Test
   void zeroReplicaCountFallsBack() {
-    final StreamThreadsCountResolver resolver = new StreamThreadsCountResolver(calculator, () -> 0);
+    final StreamThreadsCountResolver resolver = resolverWith(() -> 0, mock(AdminClient.class));
 
     final int resolved = resolver.resolve(simpleTopology(), bootstrapProperties());
 
@@ -57,8 +63,7 @@ void zeroReplicaCountFallsBack() {
 
   @Test
   void negativeReplicaCountFallsBack() {
-    final StreamThreadsCountResolver resolver =
-        new StreamThreadsCountResolver(calculator, () -> -1);
+    final StreamThreadsCountResolver resolver = resolverWith(() -> -1, mock(AdminClient.class));
 
     final int resolved = resolver.resolve(simpleTopology(), bootstrapProperties());
 
@@ -67,13 +72,15 @@ void negativeReplicaCountFallsBack() {
 
   @Test
   void delegatesToCalculatorWithConfiguredReplicas() {
-    final Topology topology = simpleTopology();
-    when(calculator.compute(eq(topology), any(), eq(8))).thenReturn(7);
-    final StreamThreadsCountResolver resolver = new StreamThreadsCountResolver(calculator, () -> 8);
+    // 30-partition source / 8 replicas -> ceil(30/8) = 4 threads. Real calculator computes this;
+    // only AdminClient (the external) is stubbed.
+    final AdminClient adminClient = mock(AdminClient.class);
+    stubPartitions(adminClient, Map.of("topic-a", 30));
+    final StreamThreadsCountResolver resolver = resolverWith(() -> 8, adminClient);
 
-    final int resolved = resolver.resolve(topology, bootstrapProperties());
+    final int resolved = resolver.resolve(simpleTopology(), bootstrapProperties());
 
-    assertEquals(7, resolved);
+    assertEquals(4, resolved);
   }
 
   @Test
@@ -94,6 +101,13 @@ void isDynamicFalseForNumericValueOrAbsent() {
     assertEquals(false, StreamThreadsCountResolver.isDynamic(Map.of()));
   }
 
+  private static StreamThreadsCountResolver resolverWith(
+      final IntSupplier replicaCountSupplier, final AdminClient adminClient) {
+    final Function<Properties, AdminClient> factory = properties -> adminClient;
+    return new StreamThreadsCountResolver(
+        new DynamicStreamThreadsCountCalculator(), replicaCountSupplier, factory);
+  }
+
   private static Topology simpleTopology() {
     final StreamsBuilder builder = new StreamsBuilder();
     builder.stream("topic-a", Consumed.with(Serdes.String(), Serdes.String()))
@@ -104,4 +118,22 @@ private static Topology simpleTopology() {
   private static Map<String, Object> bootstrapProperties() {
     return Map.of(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
   }
+
+  private static void stubPartitions(
+      final AdminClient adminClient, final Map<String, Integer> partitionsByTopic) {
+    final DescribeTopicsResult result = mock(DescribeTopicsResult.class);
+    final Map<String, KafkaFuture<TopicDescription>> futures = new HashMap<>();
+    partitionsByTopic.forEach(
+        (topic, partitionCount) -> futures.put(topic, futureFor(topic, partitionCount)));
+    when(result.topicNameValues()).thenReturn(futures);
+    when(adminClient.describeTopics(anySet())).thenReturn(result);
+  }
+
+  private static KafkaFuture<TopicDescription> futureFor(final String topic, final int partitions) {
+    final List<TopicPartitionInfo> partitionInfos =
+        IntStream.range(0, partitions)
+            .mapToObj(index -> new TopicPartitionInfo(index, Node.noNode(), List.of(), List.of()))
+            .collect(toUnmodifiableList());
+    return KafkaFuture.completedFuture(new TopicDescription(topic, false, partitionInfos));
+  }
 }