test(spanner): unflake ITBulkConnectionTest.testBulkCreateConnectionsMultiThreaded

Ensure ExecutorService in ITBulkConnectionTest.testBulkCreateConnectionsMultiThreaded is always shut down in a try-finally block if the test fails, and use JUnit assertTrue/assertNotNull assertions.

Author: sakthivelmanii

java-spanner/google-cloud-spanner/src/test/java/com/google/cloud/spanner/connection/it/ITBulkConnectionTest.java

@@ -19,6 +19,7 @@
 import static org.hamcrest.CoreMatchers.is;
 import static org.hamcrest.CoreMatchers.notNullValue;
 import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
@@ -70,24 +71,33 @@ public void testBulkCreateConnectionsSingleThreaded() {
   @Test
   public void testBulkCreateConnectionsMultiThreaded() throws Exception {
     ExecutorService executor = Executors.newFixedThreadPool(50);
-    List<Future<?>> futures = new ArrayList<>(NUMBER_OF_TEST_CONNECTIONS);
-    for (int i = 0; i < NUMBER_OF_TEST_CONNECTIONS; i++) {
-      futures.add(
-          executor.submit(
-              () -> {
-                try (ITConnection connection = createConnection()) {
-                  try (ResultSet rs = connection.executeQuery(Statement.of("select 1"))) {
-                    assertThat(rs.next(), is(true));
-                    assertThat(connection.getReadTimestamp(), is(notNullValue()));
+    try {
+      List<Future<?>> futures = new ArrayList<>(NUMBER_OF_TEST_CONNECTIONS);
+      for (int i = 0; i < NUMBER_OF_TEST_CONNECTIONS; i++) {
+        futures.add(
+            executor.submit(
+                () -> {
+                  try (ITConnection connection = createConnection()) {
+                    try (ResultSet resultSet = connection.executeQuery(Statement.of("select 1"))) {
+                      assertTrue(resultSet.next());
+                      assertNotNull(connection.getReadTimestamp());
+                    }
                   }
-                }
-                return null;
-              }));
-    }
-    executor.shutdown();
-    assertTrue(executor.awaitTermination(60L, TimeUnit.SECONDS));
-    for (Future<?> future : futures) {
-      assertNull(future.get());
+                  return null;
+                }));
+      }
+      executor.shutdown();
+      assertTrue(
+          "Executor did not terminate within timeout; active threads remain",
+          executor.awaitTermination(60L, TimeUnit.SECONDS));
+      for (Future<?> future : futures) {
+        assertNull(future.get());
+      }
+    } finally {
+      if (!executor.isTerminated()) {
+        executor.shutdownNow();
+        executor.awaitTermination(5L, TimeUnit.SECONDS);
+      }
     }
     // close Spanner instances explicitly. This method will throw an exception if there are any
     // connections still open in the pool