Update thread leak scope in ingestion IT

Signed-off-by: Varun Bharadwaj <[email protected]>
opensearch-project · Feb 3, 2025 · 3d45a2a · 3d45a2a
1 parent c20567f
commit 3d45a2a
Showing 1 changed file with 27 additions and 25 deletions.
diff --git a/...ion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/IngestFromKafkaIT.java b/...ion-kafka/src/internalClusterTest/java/org/opensearch/plugin/kafka/IngestFromKafkaIT.java
@@ -8,7 +8,7 @@
 
 package org.opensearch.plugin.kafka;
 
-import com.carrotsearch.randomizedtesting.annotations.ThreadLeakLingering;
+import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope;
 
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.Producer;
@@ -45,7 +45,7 @@
 /**
  * Integration test for Kafka ingestion
  */
-@ThreadLeakLingering(linger = 15000) // wait for container pull thread to die
+@ThreadLeakScope(ThreadLeakScope.Scope.NONE)
 public class IngestFromKafkaIT extends OpenSearchIntegTestCase {
     static final String topicName = "test";
 
@@ -75,29 +75,31 @@ public void testPluginsAreInstalled() {
     }
 
     public void testKafkaIngestion() {
-        setupKafka();
-        // create an index with ingestion source from kafka
-        createIndex(
-            "test",
-            Settings.builder()
-                .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
-                .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
-                .put("ingestion_source.type", "kafka")
-                .put("ingestion_source.pointer.init.reset", "earliest")
-                .put("ingestion_source.param.topic", "test")
-                .put("ingestion_source.param.bootstrap_servers", kafka.getBootstrapServers())
-                .build(),
-            "{\"properties\":{\"name\":{\"type\": \"text\"},\"age\":{\"type\": \"integer\"}}}}"
-        );
-
-        RangeQueryBuilder query = new RangeQueryBuilder("age").gte(21);
-        await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
-            refresh("test");
-            SearchResponse response = client().prepareSearch("test").setQuery(query).get();
-            assertThat(response.getHits().getTotalHits().value(), is(1L));
-        });
-
-        stopKafka();
+        try {
+            setupKafka();
+            // create an index with ingestion source from kafka
+            createIndex(
+                "test",
+                Settings.builder()
+                    .put(IndexMetadata.SETTING_NUMBER_OF_SHARDS, 1)
+                    .put(IndexMetadata.SETTING_NUMBER_OF_REPLICAS, 0)
+                    .put("ingestion_source.type", "kafka")
+                    .put("ingestion_source.pointer.init.reset", "earliest")
+                    .put("ingestion_source.param.topic", "test")
+                    .put("ingestion_source.param.bootstrap_servers", kafka.getBootstrapServers())
+                    .build(),
+                "{\"properties\":{\"name\":{\"type\": \"text\"},\"age\":{\"type\": \"integer\"}}}}"
+            );
+
+            RangeQueryBuilder query = new RangeQueryBuilder("age").gte(21);
+            await().atMost(10, TimeUnit.SECONDS).untilAsserted(() -> {
+                refresh("test");
+                SearchResponse response = client().prepareSearch("test").setQuery(query).get();
+                assertThat(response.getHits().getTotalHits().value(), is(1L));
+            });
+        } finally {
+            stopKafka();
+        }
     }
 
     private void setupKafka() {