fix: Address PR review comments (round 4)

- Close allocator in SpiceClient constructor if buildFlightClient() or
  initRetryers() throws, preventing off-heap memory leaks on failed
  client construction.

- Replace raw Thread usage in concurrent tests with ExecutorService
  and proper shutdownNow() cleanup, preventing thread leaks and JVM
  hangs on test timeout.

- Refactor integration tests to probe server availability in setUp()
  and gate with a boolean flag (matching TpchIntegrationTest pattern)
  instead of brittle exception message substring matching.

- Clarify README example: add comment explaining isTransportFailure()
  is application-defined and suggest which exception types to check.

Author: lukekim

.commitmsg

@@ -0,0 +1,16 @@
+fix: Address PR review comments (round 4)
+
+- Close allocator in SpiceClient constructor if buildFlightClient() or
+  initRetryers() throws, preventing off-heap memory leaks on failed
+  client construction.
+
+- Replace raw Thread usage in concurrent tests with ExecutorService
+  and proper shutdownNow() cleanup, preventing thread leaks and JVM
+  hangs on test timeout.
+
+- Refactor integration tests to probe server availability in setUp()
+  and gate with a boolean flag (matching TpchIntegrationTest pattern)
+  instead of brittle exception message substring matching.
+
+- Clarify README example: add comment explaining isTransportFailure()
+  is application-defined and suggest which exception types to check.

README.md

@@ -225,7 +225,10 @@ SpiceClient client = SpiceClient.builder()
     .withSpiceCloud()
     .build();
 
-// Long-lived usage with transport recovery
+// Long-lived usage with transport recovery.
+// isTransportFailure() is application-defined; check for
+// io.grpc.StatusRuntimeException with Status.UNAVAILABLE,
+// SSLHandshakeException, or similar transport-level errors.
 try {
     try (FlightStream stream = client.query(sql)) {
         // process results...

src/main/java/ai/spice/SpiceClient.java

@@ -223,11 +223,20 @@ public SpiceClient(String appId, String apiKey, URI flightAddress, URI httpAddre
                 : memoryLimitMB * BYTES_PER_MB;
         this.allocator = new RootAllocator(memoryLimitBytes);
 
-        // Build the Flight client (channel + auth handshake)
-        buildFlightClient();
+        try {
+            // Build the Flight client (channel + auth handshake)
+            buildFlightClient();
 
-        // Initialize cached retryers (immutable, built once)
-        initRetryers();
+            // Initialize cached retryers (immutable, built once)
+            initRetryers();
+        } catch (RuntimeException | Error e) {
+            try {
+                this.allocator.close();
+            } catch (Exception closeEx) {
+                e.addSuppressed(closeEx);
+            }
+            throw e;
+        }
 
         logger.debug("SpiceClient initialized - flightAddress={}, appId={}", this.flightAddress, this.appId);
     }

src/test/java/ai/spice/ResetTest.java

@@ -26,6 +26,9 @@ of this software and associated documentation files (the "Software"), to deal
 import java.util.ArrayList;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.arrow.flight.FlightStream;
@@ -39,6 +42,21 @@ of this software and associated documentation files (the "Software"), to deal
  */
 public class ResetTest extends TestCase {
 
+    private boolean serverAvailable = false;
+
+    @Override
+    protected void setUp() throws Exception {
+        super.setUp();
+        try (SpiceClient probe = SpiceClient.builder().build()) {
+            try (FlightStream stream = probe.query("SELECT 1")) {
+                stream.next();
+            }
+            serverAvailable = true;
+        } catch (Exception e) {
+            serverAvailable = false;
+        }
+    }
+
     // ==================== reset() Happy Path ====================
 
     /**
@@ -204,28 +222,31 @@ public void testConcurrentResetDoesNotThrow() throws Exception {
         final SpiceClient client = SpiceClient.builder().build();
         final int threadCount = 8;
         final CountDownLatch startLatch = new CountDownLatch(1);
-        final CountDownLatch doneLatch = new CountDownLatch(threadCount);
         final AtomicInteger errors = new AtomicInteger(0);
+        ExecutorService executor = Executors.newFixedThreadPool(threadCount);
 
-        for (int i = 0; i < threadCount; i++) {
-            new Thread(() -> {
-                try {
-                    startLatch.await(); // all threads start at the same time
-                    client.reset();
-                } catch (Exception e) {
-                    errors.incrementAndGet();
-                } finally {
-                    doneLatch.countDown();
-                }
-            }).start();
-        }
+        try {
+            for (int i = 0; i < threadCount; i++) {
+                executor.submit(() -> {
+                    try {
+                        startLatch.await();
+                        client.reset();
+                    } catch (Exception e) {
+                        errors.incrementAndGet();
+                    }
+                });
+            }
 
-        startLatch.countDown(); // release all threads
-        assertTrue("Concurrent reset should complete within 30s",
-                doneLatch.await(30, java.util.concurrent.TimeUnit.SECONDS));
+            startLatch.countDown(); // release all threads
+            executor.shutdown();
+            assertTrue("Concurrent reset should complete within 30s",
+                    executor.awaitTermination(30, TimeUnit.SECONDS));
 
-        assertEquals("No threads should have encountered errors", 0, errors.get());
-        client.close();
+            assertEquals("No threads should have encountered errors", 0, errors.get());
+        } finally {
+            executor.shutdownNow();
+            client.close();
+        }
     }
 
     /**
@@ -236,51 +257,52 @@ public void testConcurrentResetAndQuery() throws Exception {
         final SpiceClient client = SpiceClient.builder().build();
         final int iterations = 5;
         final CountDownLatch startLatch = new CountDownLatch(1);
-        final CountDownLatch doneLatch = new CountDownLatch(2);
         final List<Throwable> unexpectedErrors = new ArrayList<>();
+        ExecutorService executor = Executors.newFixedThreadPool(2);
 
-        // Thread 1: repeated resets
-        new Thread(() -> {
-            try {
-                startLatch.await();
-                for (int i = 0; i < iterations; i++) {
-                    client.reset();
-                    Thread.sleep(10);
+        try {
+            // Thread 1: repeated resets
+            executor.submit(() -> {
+                try {
+                    startLatch.await();
+                    for (int i = 0; i < iterations; i++) {
+                        client.reset();
+                        Thread.sleep(10);
+                    }
+                } catch (InterruptedException ignored) {
                 }
-            } catch (InterruptedException ignored) {
-            } finally {
-                doneLatch.countDown();
-            }
-        }).start();
+            });
 
-        // Thread 2: repeated queries
-        new Thread(() -> {
-            try {
-                startLatch.await();
-                for (int i = 0; i < iterations; i++) {
-                    try {
-                        FlightStream stream = client.query("SELECT 1");
-                        stream.close();
-                    } catch (NullPointerException e) {
-                        unexpectedErrors.add(e);
-                    } catch (Exception e) {
-                        // Connection errors are expected
+            // Thread 2: repeated queries
+            executor.submit(() -> {
+                try {
+                    startLatch.await();
+                    for (int i = 0; i < iterations; i++) {
+                        try {
+                            FlightStream stream = client.query("SELECT 1");
+                            stream.close();
+                        } catch (NullPointerException e) {
+                            unexpectedErrors.add(e);
+                        } catch (Exception e) {
+                            // Connection errors are expected
+                        }
+                        Thread.sleep(10);
                     }
-                    Thread.sleep(10);
+                } catch (InterruptedException ignored) {
                 }
-            } catch (InterruptedException ignored) {
-            } finally {
-                doneLatch.countDown();
-            }
-        }).start();
+            });
 
-        startLatch.countDown();
-        assertTrue("Concurrent reset+query should complete within 30s",
-                doneLatch.await(30, java.util.concurrent.TimeUnit.SECONDS));
+            startLatch.countDown();
+            executor.shutdown();
+            assertTrue("Concurrent reset+query should complete within 30s",
+                    executor.awaitTermination(30, TimeUnit.SECONDS));
 
-        assertTrue("Should not get NullPointerException during concurrent reset+query: " + unexpectedErrors,
-                unexpectedErrors.isEmpty());
-        client.close();
+            assertTrue("Should not get NullPointerException during concurrent reset+query: " + unexpectedErrors,
+                    unexpectedErrors.isEmpty());
+        } finally {
+            executor.shutdownNow();
+            client.close();
+        }
     }
 
     // ==================== Construction / DNS / Keep-alive ====================
@@ -391,6 +413,8 @@ public void testResetWithCustomConfig() throws Exception {
      * reset() followed by query() actually returns data.
      */
     public void testResetThenQueryIntegration() throws Exception {
+        if (!serverAvailable) return;
+
         try (SpiceClient client = SpiceClient.builder().build()) {
             // First query (establishes connection)
             try (FlightStream stream1 = client.query(
@@ -414,14 +438,6 @@ public void testResetThenQueryIntegration() throws Exception {
                 }
                 assertEquals("Second query after reset should return 2 rows", 2, rows2);
             }
-        } catch (Exception e) {
-            // Skip if no local runtime or TPC-H data available
-            String msg = e.getMessage() != null ? e.getMessage().toLowerCase() : "";
-            if (msg.contains("unavailable") || msg.contains("connection refused")
-                    || msg.contains("not found") || msg.contains("io exception")) {
-                return;
-            }
-            fail("Unexpected error: " + e.getMessage());
         }
     }
 
@@ -430,6 +446,8 @@ public void testResetThenQueryIntegration() throws Exception {
      * reset() followed by queryWithParams() actually returns data.
      */
     public void testResetThenQueryWithParamsIntegration() throws Exception {
+        if (!serverAvailable) return;
+
         try (SpiceClient client = SpiceClient.builder().build()) {
             // First query
             try (ArrowReader reader1 = client.queryWithParams(
@@ -455,13 +473,6 @@ public void testResetThenQueryWithParamsIntegration() throws Exception {
                 }
                 assertTrue("Second parameterized query after reset should return rows", rows2 > 0);
             }
-        } catch (Exception e) {
-            String msg = e.getMessage() != null ? e.getMessage().toLowerCase() : "";
-            if (msg.contains("unavailable") || msg.contains("connection refused")
-                    || msg.contains("not found") || msg.contains("io exception")) {
-                return;
-            }
-            fail("Unexpected error: " + e.getMessage());
         }
     }
 }