Fix snappy related Kafka errors and relax thread leak checks

Signed-off-by: Varun Bharadwaj <[email protected]>

Author: varunbharadwaj

plugins/ingestion-kafka/build.gradle

@@ -21,12 +21,14 @@ versions << [
   'docker': '3.3.6',
   'testcontainers': '1.19.7',
   'ducttape': '1.0.8',
+  'snappy': '1.1.10.7',
 ]
 
 dependencies {
   // kafka
   api "org.slf4j:slf4j-api:${versions.slf4j}"
   api "org.apache.kafka:kafka-clients:${versions.kafka}"
+  api "org.xerial.snappy:snappy-java:${versions.snappy}"
 
   // test
   testImplementation "com.github.docker-java:docker-java-api:${versions.docker}"

plugins/ingestion-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() {

plugins/ingestion-kafka/src/main/plugin-metadata/plugin-security.policy

@@ -17,5 +17,8 @@ grant {
   // Allow host/ip name service lookups
   permission java.net.SocketPermission "*", "connect";
   permission java.net.SocketPermission "*", "resolve";
-};
 
+  // Needed for Kafka consumer to load native snappy library
+  permission java.lang.RuntimePermission "loadLibrary.*";
+  permission java.io.FilePermission "${java.io.tmpdir}${/}snappy-*", "read,write";
+};