Fix failing tests; add no-op close implementation

Author: mtdowling

client/client-core/src/main/java/software/amazon/smithy/java/client/core/ClientTransport.java

@@ -6,6 +6,7 @@
 package software.amazon.smithy.java.client.core;
 
 import java.io.Closeable;
+import java.io.IOException;
 import java.net.ConnectException;
 import java.net.ProtocolException;
 import java.net.SocketException;
@@ -47,6 +48,14 @@ public interface ClientTransport<RequestT, ResponseT> extends Closeable {
      */
     MessageExchange<RequestT, ResponseT> messageExchange();
 
+    /**
+     * {@inheritDoc}
+     *
+     * <p>Default implementation is a no-op.
+     */
+    @Override
+    default void close() throws IOException {}
+
     /**
      * Remaps a thrown exception to an appropriate {@link TransportException} or {@link CallException}.
      *

http/http-client/src/it/java/software/amazon/smithy/java/http/client/it/h2/TrailerHeadersHttp2Test.java

@@ -7,7 +7,6 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.fail;
 
 import java.util.Map;
 import org.junit.jupiter.api.Test;
@@ -43,28 +42,15 @@ protected HttpConnectionPoolBuilder configurePool(HttpConnectionPoolBuilder buil
 
     @Test
     void readsResponseWithTrailers() throws Exception {
-        // Retry up to 5 times due to timing sensitivity in H2 frame ordering
-        Exception lastException = null;
-        for (int attempt = 0; attempt < 5; attempt++) {
-            try {
-                var request = plainTextRequest(HttpVersion.HTTP_2, "");
-                try (var exchange = client.newExchange(request)) {
-                    var body = new String(exchange.responseBody().readAllBytes());
-                    assertEquals(RESPONSE_CONTENTS, body);
-
-                    var trailers = exchange.responseTrailerHeaders();
-                    assertNotNull(trailers, "Should have trailer headers");
-                    assertEquals("abc123", trailers.firstValue("x-checksum"));
-                    assertEquals("req-456", trailers.firstValue("x-request-id"));
-                }
-                return; // Success
-            } catch (Exception e) {
-                lastException = e;
-                // Recreate client/server for retry
-                tearDown();
-                setUp();
-            }
+        var request = plainTextRequest(HttpVersion.HTTP_2, "");
+        try (var exchange = client.newExchange(request)) {
+            var body = new String(exchange.responseBody().readAllBytes());
+            assertEquals(RESPONSE_CONTENTS, body);
+
+            var trailers = exchange.responseTrailerHeaders();
+            assertNotNull(trailers, "Should have trailer headers");
+            assertEquals("abc123", trailers.firstValue("x-checksum"));
+            assertEquals("req-456", trailers.firstValue("x-request-id"));
         }
-        fail("Test failed after 5 attempts", lastException);
     }
 }

http/http-client/src/main/java/software/amazon/smithy/java/http/client/h2/H2Exchange.java

@@ -83,8 +83,8 @@ public final class H2Exchange implements HttpExchange {
     private final H2StreamState state = new H2StreamState();
 
     // Pending headers from reader thread (protected by dataLock)
-    private List<String> pendingHeaders;
-    private boolean pendingHeadersEndStream;
+    private record PendingHeadersEvent(List<String> fields, boolean endStream) {}
+    private final ArrayDeque<PendingHeadersEvent> pendingHeadersQueue = new ArrayDeque<>();
 
     // === Data chunk queue ===
     // Queue of DataChunks received from reader thread. Each chunk contains one DATA frame payload.
@@ -352,8 +352,7 @@ void returnBuffer(byte[] buffer) {
     void deliverHeaders(List<String> fields, boolean endStream) {
         dataLock.lock();
         try {
-            pendingHeaders = fields;
-            pendingHeadersEndStream = endStream;
+            pendingHeadersQueue.add(new PendingHeadersEvent(fields, endStream));
         } finally {
             dataLock.unlock();
         }
@@ -496,11 +495,9 @@ int drainChunks(DataChunk[] dest, int maxChunks) throws IOException {
             // Wait for data, EOF, or error
             while (dataQueue.isEmpty() && state.getReadState() == RS_READING) {
                 // Check for pending trailers
-                if (pendingHeaders != null) {
-                    List<String> fields = pendingHeaders;
-                    boolean endStream = pendingHeadersEndStream;
-                    pendingHeaders = null;
-                    handleHeadersEvent(fields, endStream);
+                PendingHeadersEvent headerEvent = pendingHeadersQueue.poll();
+                if (headerEvent != null) {
+                    handleHeadersEvent(headerEvent.fields(), headerEvent.endStream());
                     if (state.getReadState() == RS_DONE) {
                         break;
                     }
@@ -726,7 +723,7 @@ private void awaitEvent() throws IOException {
         try {
             // Wait for headers, error, or data (which also signals)
             int rs;
-            while (pendingHeaders == null && (rs = state.getReadState()) != RS_ERROR && rs != RS_DONE) {
+            while (pendingHeadersQueue.isEmpty() && (rs = state.getReadState()) != RS_ERROR && rs != RS_DONE) {
                 // Wait using lock-free signaling
                 waitingThread = Thread.currentThread();
                 dataLock.unlock();
@@ -764,13 +761,10 @@ private void readResponseHeaders() throws IOException {
 
             dataLock.lock();
             try {
-                if (pendingHeaders != null) {
-                    List<String> fields = pendingHeaders;
-                    boolean endStream = pendingHeadersEndStream;
-                    pendingHeaders = null; // Consume the headers
-
+                PendingHeadersEvent headerEvent = pendingHeadersQueue.poll();
+                if (headerEvent != null) {
                     // Process headers (can throw)
-                    handleHeadersEvent(fields, endStream);
+                    handleHeadersEvent(headerEvent.fields(), headerEvent.endStream());
                 } else if (state.getReadState() == RS_DONE) {
                     throw new IOException("Stream ended before response headers received");
                 }

io/src/test/java/software/amazon/smithy/java/io/datastream/PublisherDataStreamTest.java

@@ -27,13 +27,31 @@ void writeTo() throws IOException {
         var ds = DataStream.ofPublisher(publisher, null, -1);
         var out = new ByteArrayOutputStream();
 
-        Thread.startVirtualThread(() -> {
-            publisher.submit(ByteBuffer.wrap(chunk1));
-            publisher.submit(ByteBuffer.wrap(chunk2));
-            publisher.close();
+        // Run writeTo on a virtual thread so it subscribes to the publisher
+        // before items are submitted, avoiding a race where close() fires
+        // before the subscription is established.
+        var writeThread = Thread.startVirtualThread(() -> {
+            try {
+                ds.writeTo(out);
+            } catch (IOException e) {
+                throw new RuntimeException(e);
+            }
         });
 
-        ds.writeTo(out);
+        // Wait for writeTo's subscriber to be registered.
+        while (publisher.getNumberOfSubscribers() < 1) {
+            Thread.onSpinWait();
+        }
+
+        publisher.submit(ByteBuffer.wrap(chunk1));
+        publisher.submit(ByteBuffer.wrap(chunk2));
+        publisher.close();
+
+        try {
+            writeThread.join();
+        } catch (InterruptedException e) {
+            throw new RuntimeException(e);
+        }
 
         assertArrayEquals("hello world".getBytes(StandardCharsets.UTF_8), out.toByteArray());
     }