KAFKA-15816: Fix leaked sockets in streams tests

streams/src/test/java/org/apache/kafka/streams/integration/IQv2IntegrationTest.java

@@ -418,6 +418,8 @@ public String metricsScope() {
             })
         );
 
+        // Discard the basic streams and replace with test-specific topology
+        kafkaStreams.close();
         kafkaStreams = new KafkaStreams(builder.build(), streamsConfiguration(testInfo));
         kafkaStreams.cleanUp();
 

streams/src/test/java/org/apache/kafka/streams/integration/MetricsReporterIntegrationTest.java

@@ -119,12 +119,12 @@ public void shouldBeAbleToProvideInitialMetricValueToMetricsReporter() {
         builder.stream(STREAM_INPUT, Consumed.with(Serdes.Integer(), Serdes.String()))
                 .to(STREAM_OUTPUT, Produced.with(Serdes.Integer(), Serdes.String()));
         final Topology topology = builder.build();
-        final KafkaStreams kafkaStreams = new KafkaStreams(topology, streamsConfiguration);
-
-        kafkaStreams.metrics().keySet().forEach(metricName -> {
-            final Object initialMetricValue = METRIC_NAME_TO_INITIAL_VALUE.get(metricName.name());
-            assertThat(initialMetricValue, notNullValue());
-        });
+        try (KafkaStreams kafkaStreams = new KafkaStreams(topology, streamsConfiguration)) {
+            kafkaStreams.metrics().keySet().forEach(metricName -> {
+                final Object initialMetricValue = METRIC_NAME_TO_INITIAL_VALUE.get(metricName.name());
+                assertThat(initialMetricValue, notNullValue());
+            });
+        }
     }
 
 }

streams/src/test/java/org/apache/kafka/streams/integration/NamedTopologyIntegrationTest.java

@@ -715,6 +715,7 @@ public void shouldAddToEmptyInitialTopologyRemoveResetOffsetsThenAddSameNamedTop
 
     @Test
     public void shouldContinueProcessingOtherTopologiesWhenNewTopologyHasMissingInputTopics() throws Exception {
+        // This test leaks sockets due to KAFKA-15834
         try {
             CLUSTER.createTopic(EXISTING_STREAM, 2, 1);
             produceToInputTopics(EXISTING_STREAM, STANDARD_INPUT_DATA);
@@ -839,6 +840,8 @@ public void shouldBackOffTaskAndEmitDataWithinSameTopology() throws Exception {
             props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.IntegerSerde.class);
             props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.StringSerde.class);
 
+            // Discard the pre-created streams and replace with test-specific topology
+            streams.close();
             streams = new KafkaStreamsNamedTopologyWrapper(props);
             streams.setUncaughtExceptionHandler(exception -> StreamThreadExceptionResponse.REPLACE_THREAD);
 

streams/src/test/java/org/apache/kafka/streams/integration/PurgeRepartitionTopicIntegrationTest.java

@@ -181,6 +181,9 @@ public void setup() {
 
     @AfterEach
     public void shutdown() {
+        if (adminClient != null) {
+            adminClient.close();
+        }
         if (kafkaStreams != null) {
             kafkaStreams.close(Duration.ofSeconds(30));
         }