Verification that the threads are properly shutdown in the endOfFragmentTest (#274)

* Verification that the threads are properly shutdown in the endOfFragmentTest

* add polling mechanism to ensure all threads are cleaned up

* added extra logging for thread clean up process in EndOfFragmentIntegTest

---------

Co-authored-by: Yuqi Huang <[email protected]>

Author: sirknightj

src/test/java/com/amazonaws/kinesisvideo/common/EndOfFragmentIntegTest.java

@@ -24,6 +24,7 @@
 import java.nio.ByteBuffer;
 import java.time.Duration;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
@@ -88,6 +89,20 @@ public class EndOfFragmentIntegTest extends ProducerTestBase {
     private static final int NUMBER_OF_FRAMES_TO_STREAM = 10;
     private static final int FPS = 5;
     private static final int KEYFRAME_INTERVAL = 5;
+    private static final int SHUTDOWN_TIMEOUT_MS = 5000; //total time allowed for clean up
+    private static final int INTERVAL_MS = 100; //time interval between each clean up
+
+    /**
+     * Names of the threads that exist before the test.
+     */
+    private List<String> threadsBefore;
+
+    /**
+     * List of thread names to ignore.
+     */
+    private List<String> threadsToIgnore = Arrays.asList(
+            "java-sdk-http-connection-reaper" // Owned by AWS SDK Java
+    );
 
     /**
      * List of streams created during tests that need to be cleaned up.
@@ -113,6 +128,14 @@ public void setUp() {
         assumeTrue("You need to increase the number of streams the client is configured for before starting the test!",
                 NUMBER_OF_STREAMS_PER_ITERATION <= NUMBER_OF_STREAMS);
 
+        // Capture baseline thread state
+        this.threadsBefore = Thread.getAllStackTraces().keySet()
+                .stream()
+                .map(Thread::getName)
+                .collect(Collectors.toList());
+
+        threadsBefore.sort(String.CASE_INSENSITIVE_ORDER);
+
         createProducer();
     }
 
@@ -150,6 +173,44 @@ public void tearDown() {
         this.createdStreams.clear();
 
         assertFalse("An exception happened during cleanup!", failure);
+
+        freeProducer();
+        threadsBefore.removeAll(threadsToIgnore);
+        for (int i = 0; i < SHUTDOWN_TIMEOUT_MS; i += INTERVAL_MS) {
+            final List<String> threadsNow = Thread.getAllStackTraces().keySet()
+                    .stream()
+                    .map(Thread::getName)
+                    .collect(Collectors.toList());
+
+            threadsNow.sort(String.CASE_INSENSITIVE_ORDER);
+            threadsNow.removeAll(threadsToIgnore);
+            log.info("Cleanup iteration {}/() ms - Current thread count: {}, Expected: {}",
+                    i, SHUTDOWN_TIMEOUT_MS, threadsNow.size(), this.threadsBefore.size());
+            
+            if (threadsNow.equals(this.threadsBefore)) {
+                break; // threads are cleaned up
+            } else {
+                //if threads are not clearned up yet
+                List<String> extraThreads = new ArrayList<>(threadsNow);
+                extraThreads.removeAll(this.threadsBefore);
+                if(!extraThreads.isEmpty()) {
+                    log.warn("extra threads are still running: {}", extraThreads);
+                }
+            }
+
+            if (i + INTERVAL_MS >= SHUTDOWN_TIMEOUT_MS) {
+                //time has exceeded shutdown timeout
+                log.error("Expected threads: {}", this.threadsBefore);
+                log.error("Current threads: {}", threadsNow);
+                fail("Timeout waiting for threads to be cleaned up properly");               
+            }
+
+            try {
+                Thread.sleep(INTERVAL_MS);
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            }
+        }
     }
 
     // Using this as a way to repeat the test multiple times
@@ -278,6 +339,8 @@ public void test_When_StoppingNearIntermittentProducer_Then_StreamingSuccessfull
                 .count();
 
         assertTrue("Didn't receive any PERSISTED ACKs. Received: " + this.receivedFragmentAcks_, persistedAcksCount > 0);
+
+        executorService.shutdownNow();
     }
 
     /**

src/test/java/com/amazonaws/kinesisvideo/common/ProducerTestBase.java

@@ -6,8 +6,10 @@
 import java.util.Optional;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 import static com.amazonaws.kinesisvideo.internal.producer.jni.NativeKinesisVideoProducerJni.PRODUCER_NATIVE_LIBRARY_NAME;
+import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 
 import com.amazonaws.auth.AWSCredentialsProvider;
@@ -179,6 +181,24 @@ protected void createProducer(DeviceInfo deviceInfo) {
         }
     }
 
+    protected void freeProducer() {
+        try {
+            this.kinesisVideoProducer.free();
+        } catch (ProducerException e) {
+            log.error("Failed to free the producer", e);
+            fail(e.getMessage());
+        }
+
+        this.executor.shutdownNow();
+        try {
+            assertTrue("Didn't shutdown the executor service in time!",
+                    this.executor.awaitTermination(5, TimeUnit.SECONDS));
+        } catch (final InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.error("Interrupted while waiting for executor to terminate", e);
+        }
+    }
+
     /**
      * This method is used to create a stream with the specified information using the producer created as a part of
      * the createProducer method
@@ -306,6 +326,8 @@ protected void prepareStream(final String streamName) {
                 }
             }
         }
+
+        kvs.shutdown();
     }
 
     /**