diff --git a/flash/src/main/java/dev/relism/flash/extension/FlashConfiguration.java b/flash/src/main/java/dev/relism/flash/extension/FlashConfiguration.java index 455f9f3..b92d40c 100644 --- a/flash/src/main/java/dev/relism/flash/extension/FlashConfiguration.java +++ b/flash/src/main/java/dev/relism/flash/extension/FlashConfiguration.java @@ -88,6 +88,17 @@ public class FlashConfiguration { */ @Builder.Default int shutdownDrainTimeoutMs = 15_000; + /** + * Maximum connections admitted across all listeners before new connections are closed + * immediately at accept time, before any per-connection state (TLS handshake, protocol + * negotiation, HPACK tables, buffers) is set up. Defaults to an auto-scaled budget based on the + * JVM's max heap ({@link dev.relism.flash.transport.TransportLimits#defaultMaxConnections()}), + * so a connection flood cannot exhaust the heap out of the box. Set explicitly if you know your + * deployment's real capacity, or to {@code 0} to disable the check entirely (unlimited). + */ + @Builder.Default int maxConnections = + dev.relism.flash.transport.TransportLimits.defaultMaxConnections(); + /** Whether TLS listeners advertise HTTP/2 through ALPN. */ @Builder.Default boolean http2Enabled = false; diff --git a/flash/src/main/java/dev/relism/flash/http1/Http1ResponseWriter.java b/flash/src/main/java/dev/relism/flash/http1/Http1ResponseWriter.java index fe0b68f..305aac2 100644 --- a/flash/src/main/java/dev/relism/flash/http1/Http1ResponseWriter.java +++ b/flash/src/main/java/dev/relism/flash/http1/Http1ResponseWriter.java @@ -120,7 +120,7 @@ public final class Http1ResponseWriter { out.write(head.array(), 0, head.length()); if (suppressBody) return; if (response.isStreaming()) { - writeChunked(out, response.getStream(), response, scratch); + writeChunkedAndClose(out, response, scratch); } else { byte[] body = response.getBody(); if (body != null && body.length != 0) { @@ -145,7 +145,7 @@ public final class Http1ResponseWriter { head.writeBytes(keepAlive ? CONNECTION_KEEPALIVE : CONNECTION_CLOSE); head.writeBytes(CRLF); out.write(head.array(), 0, head.length()); - if (!suppressBody) relay(response.getStream(), out, scratch); + if (!suppressBody) relayAndClose(response.getStream(), out, scratch); } else { head.writeBytes(TRANSFER_CHUNKED); head.writeBytes(keepAlive ? CONNECTION_KEEPALIVE : CONNECTION_CLOSE); @@ -154,7 +154,34 @@ public final class Http1ResponseWriter { // A HEAD response still declares the Transfer-Encoding GET would have used (RFC // 9110 §9.3.2) but writes zero body bytes — not even the final-chunk marker, since // there is no chunk framing at all for a message with no body. - if (!suppressBody) writeChunked(out, response.getStream(), response, scratch); + if (!suppressBody) writeChunkedAndClose(out, response, scratch); + } + } + + /** + * Closes the handler's stream on every exit — clean EOF or a write failure partway through + * (e.g. the client disconnected mid-transfer). Without this, a handler whose stream only + * releases a held resource (a pooled backend connection, say) from {@code close()} — not from + * observing EOF on a {@code read()} that a downstream write failure means it never reaches — + * leaks that resource for as long as the JVM takes to finalize it. A well-behaved stream's + * {@code close()} must already be idempotent (Java's own contract for {@link InputStream}), so + * this costs nothing extra on the ordinary clean-EOF path. + */ + private static void relayAndClose(InputStream in, OutputStream out, ConnectionScratch scratch) + throws IOException { + try { + relay(in, out, scratch); + } finally { + in.close(); + } + } + + private static void writeChunkedAndClose(OutputStream out, Response response, ConnectionScratch scratch) + throws IOException { + try { + writeChunked(out, response.getStream(), response, scratch); + } finally { + response.getStream().close(); } } diff --git a/flash/src/main/java/dev/relism/flash/http2/Http2Limits.java b/flash/src/main/java/dev/relism/flash/http2/Http2Limits.java index 1fc31cd..ad11118 100644 --- a/flash/src/main/java/dev/relism/flash/http2/Http2Limits.java +++ b/flash/src/main/java/dev/relism/flash/http2/Http2Limits.java @@ -70,8 +70,19 @@ public final class Http2Limits { * companion bound to {@link #MAX_RESET_STREAMS_PER_INTERVAL}: Rapid Reset defences that only * count resets can still be bypassed by a peer that creates streams fast enough that the reset * counter never saturates within any single window boundary. + * + *

Matches {@link #MAX_STREAMS_PER_CONNECTION}'s lifetime budget by design: a connection may + * not create more streams in one rolling burst window than it is ever allowed to create in its + * whole lifetime. An earlier value of 400 (40/s) measured the RST_STREAM flood attack this bound + * exists for, but also rejected ordinary high-concurrency multiplexed clients well below the + * throughput a hardened server is expected to sustain — h2load's default light-load pattern (10 + * connections, 10 concurrent streams each) alone drives multiple thousands of legitimate stream + * creations per connection per second on a fast peer, which 400/10s cannot distinguish from + * abuse. The RST_STREAM-rate counter above measures the actual CVE-2023-44487 signature (resets, + * not creates); this bound only needs to catch a peer creating streams fast enough to dodge that + * counter, which a much higher ceiling still does. */ - public static final int MAX_STREAMS_CREATED_PER_INTERVAL = 400; + public static final int MAX_STREAMS_CREATED_PER_INTERVAL = 100_000; /** Maximum SETTINGS frames accepted within one abuse-rate interval. */ public static final int MAX_SETTINGS_PER_INTERVAL = 100; diff --git a/flash/src/main/java/dev/relism/flash/http2/message/Http2HeaderMap.java b/flash/src/main/java/dev/relism/flash/http2/message/Http2HeaderMap.java index 645616a..2853f86 100644 --- a/flash/src/main/java/dev/relism/flash/http2/message/Http2HeaderMap.java +++ b/flash/src/main/java/dev/relism/flash/http2/message/Http2HeaderMap.java @@ -18,7 +18,7 @@ public final class Http2HeaderMap implements HeaderView { private HpackHeaderBlock block; private PseudoHeaders pseudoHeaders; private int viewCursor; - private int regularCount; + private int regularCount = -1; public Http2HeaderMap() { for (int i = 0; i < views.length; i++) views[i] = new PooledSlice(); @@ -28,11 +28,7 @@ public final class Http2HeaderMap implements HeaderView { this.block = block; this.pseudoHeaders = pseudoHeaders; viewCursor = 0; - regularCount = 0; - for (int i = 0; i < block.count(); i++) { - block.get(i, scanName, scanValue); - if (scanName.byteAt(0) != ':') regularCount++; - } + regularCount = -1; } public void reset(HpackHeaderBlock block) { @@ -67,11 +63,16 @@ public final class Http2HeaderMap implements HeaderView { @Override public List all() { - List result = new ArrayList<>(regularCount); + List result = new ArrayList<>(regularCount < 0 ? block.count() : regularCount); + int found = 0; for (int i = 0; i < block.count(); i++) { block.get(i, scanName, scanValue); - if (scanName.byteAt(0) != ':') result.add(string(scanValue)); + if (scanName.byteAt(0) != ':') { + result.add(string(scanValue)); + found++; + } } + regularCount = found; return result; } @@ -94,6 +95,14 @@ public final class Http2HeaderMap implements HeaderView { @Override public int count() { + if (regularCount < 0) { + int found = 0; + for (int i = 0; i < block.count(); i++) { + block.get(i, scanName, scanValue); + if (scanName.byteAt(0) != ':') found++; + } + regularCount = found; + } return regularCount; } diff --git a/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamState.java b/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamState.java index d682dd5..29e3d85 100644 --- a/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamState.java +++ b/flash/src/main/java/dev/relism/flash/http2/stream/Http2StreamState.java @@ -26,6 +26,8 @@ public enum Http2StreamState { private static final byte ERROR = -1; private static final byte[][] TRANSITIONS = buildTransitions(); + // enum values() clones its backing array on every call; this is read-only and shared safely. + private static final Http2StreamState[] VALUES = values(); public Http2StreamState transition(int streamId, Event event) { int next = TRANSITIONS[ordinal()][event.ordinal()]; @@ -33,7 +35,7 @@ public enum Http2StreamState { throw new Http2StreamException( streamId, errorFor(event), "invalid stream transition " + this + " + " + event); } - return values()[next]; + return VALUES[next]; } private Http2ErrorCode errorFor(Event event) { diff --git a/flash/src/main/java/dev/relism/flash/transport/ConnectionRunner.java b/flash/src/main/java/dev/relism/flash/transport/ConnectionRunner.java index 2ccdb9e..d1e24e5 100644 --- a/flash/src/main/java/dev/relism/flash/transport/ConnectionRunner.java +++ b/flash/src/main/java/dev/relism/flash/transport/ConnectionRunner.java @@ -61,21 +61,38 @@ public final class ConnectionRunner { * Submits {@code socket} to the virtual-thread executor for full connection handling. {@code * stopped} is threaded through to the eventual {@link ConnectionContext} so the protocol * implementation can observe an in-progress graceful shutdown. + * + *

Rejects before any per-connection state exists — no TLS handshake, no protocol + * negotiation, no HPACK tables — once {@code activeSockets} reaches {@link + * FlashConfiguration#getMaxConnections()}. This is an approximate check (accept runs on up to + * {@link TransportTuning#ACCEPT_THREADS} concurrent threads, so a burst can briefly land a few + * connections past the limit), not an atomic guarantee; it only needs to bound worst-case + * growth, not enforce an exact count. */ public void accept(Socket socket, BooleanSupplier stopped) { + int max = configuration.getMaxConnections(); + if (max > 0 && activeSockets.size() >= max) { + closeQuietly(socket); + return; + } + activeSockets.add(socket); try { executorService.submit(() -> handle(socket, stopped)); } catch (RejectedExecutionException ignored) { - try { - socket.close(); - } catch (IOException e) { - log.debug("Error closing socket on shutdown", e); - } + activeSockets.remove(socket); + closeQuietly(socket); + } + } + + private static void closeQuietly(Socket socket) { + try { + socket.close(); + } catch (IOException e) { + log.debug("Error closing socket on shutdown", e); } } private void handle(Socket socket, BooleanSupplier stopped) { - activeSockets.add(socket); ConnectionScratch scratch = scratchPool.acquire(); try (socket; OutputStream out = new BufferedOutputStream(socket.getOutputStream())) { diff --git a/flash/src/main/java/dev/relism/flash/transport/TransportLimits.java b/flash/src/main/java/dev/relism/flash/transport/TransportLimits.java new file mode 100644 index 0000000..b3fa5fa --- /dev/null +++ b/flash/src/main/java/dev/relism/flash/transport/TransportLimits.java @@ -0,0 +1,50 @@ +package dev.relism.flash.transport; + +/** + * Transport-level bound on concurrent connections, enforced by {@link ConnectionRunner} before + * any per-connection state (TLS handshake, protocol negotiation, HPACK tables, buffers) is set + * up. Unlike {@link dev.relism.flash.http2.Http2Limits} (bounds on what one already-admitted + * connection may do), this bounds how many connections are admitted at all — the guard a stress + * test found completely absent: {@code AcceptLoop} accepted unconditionally, so a connection + * flood ran the JVM out of heap rather than being turned away. + */ +public final class TransportLimits { + + private TransportLimits() {} + + /** + * Deliberately conservative estimate of one connection's worst-case retained heap (HPACK + * tables, stream table, in-flight response batches, up to {@code + * Http2Limits#MAX_CONCURRENT_STREAMS} concurrent streams), used only to size {@link + * #defaultMaxConnections()}'s auto-scaled budget — not an enforced per-connection cap. + * + *

Not a precise per-byte accounting. A stress test on this codebase (h2load, 20 concurrent + * HTTP/2 streams per connection) observed {@code OutOfMemoryError} somewhere between 200 and + * 400 concurrent connections on a 1.5 GiB heap. This constant is chosen so {@link + * #defaultMaxConnections()} lands comfortably below that observed floor (~150 connections at + * 1.5 GiB) rather than hugging it. A heap-dump-derived precise figure is a natural follow-up; + * until then this trades some throughput headroom for a real safety margin. + */ + static final long ASSUMED_WORST_CASE_BYTES_PER_CONNECTION = 5L * 1024 * 1024; + + /** + * Fraction of the JVM's max heap set aside for connection-admission accounting; the rest is + * left for GC headroom, response buffers, and everything else the server needs. + */ + static final double HEAP_FRACTION_FOR_CONNECTIONS = 0.5; + + /** Floor so a tiny heap (dev/test containers) still gets a usable, non-degenerate limit. */ + static final int MIN_MAX_CONNECTIONS = 64; + + /** + * Auto-scaled default for {@code FlashConfiguration#getMaxConnections()}. Computed from {@link + * Runtime#maxMemory()} so the same default protects a 256 MiB container and an 8 GiB one + * without operator input; set {@code maxConnections} explicitly to override it, or to {@code 0} + * to disable the check (unlimited — the behavior every version before this had unconditionally). + */ + public static int defaultMaxConnections() { + long heapBudget = (long) (Runtime.getRuntime().maxMemory() * HEAP_FRACTION_FOR_CONNECTIONS); + long computed = heapBudget / ASSUMED_WORST_CASE_BYTES_PER_CONNECTION; + return (int) Math.max(MIN_MAX_CONNECTIONS, Math.min(Integer.MAX_VALUE, computed)); + } +} diff --git a/flash/src/test/java/dev/relism/flash/http1/Http1ResponseWriterTest.java b/flash/src/test/java/dev/relism/flash/http1/Http1ResponseWriterTest.java index be0836d..fd885e4 100644 --- a/flash/src/test/java/dev/relism/flash/http1/Http1ResponseWriterTest.java +++ b/flash/src/test/java/dev/relism/flash/http1/Http1ResponseWriterTest.java @@ -7,8 +7,11 @@ import dev.relism.flash.transport.ConnectionScratch; import dev.relism.flash.transport.ScratchPool; import org.junit.jupiter.api.Test; +import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; import java.io.IOException; +import java.io.InputStream; +import java.io.OutputStream; import java.nio.charset.StandardCharsets; import java.util.Arrays; @@ -177,4 +180,74 @@ class Http1ResponseWriterTest { assertEquals(1, out.arrayWriteCalls); assertFalse(out.sink.toString(StandardCharsets.UTF_8).contains("hello world")); } + + // --- Streaming body close-on-every-exit ----------------------------------------- + + /** Tracks whether {@code close()} was called, regardless of how the stream was read. */ + private static final class TrackingInputStream extends ByteArrayInputStream { + boolean closed; + + TrackingInputStream(byte[] buf) { + super(buf); + } + + @Override + public void close() throws IOException { + closed = true; + super.close(); + } + } + + /** Simulates a client disconnecting mid-transfer: the Nth write() call throws. */ + private static final class FailingOutputStream extends OutputStream { + private final int failAfterCalls; + private int calls; + + FailingOutputStream(int failAfterCalls) { + this.failAfterCalls = failAfterCalls; + } + + @Override public void write(int b) {} + + @Override + public void write(byte[] b, int off, int len) throws IOException { + calls++; + if (calls > failAfterCalls) throw new IOException("simulated client disconnect"); + } + } + + @Test + void chunkedStreamingBody_isClosed_onCleanCompletion() throws IOException { + TrackingInputStream stream = new TrackingInputStream("hello world".getBytes(StandardCharsets.UTF_8)); + Response response = new Response(200, ContentType.TEXT_PLAIN).chunked(stream); + Http1ResponseWriter.writeResponse(new ByteArrayOutputStream(), response, HttpMethod.GET, true, false, scratch()); + + assertTrue(stream.closed, "a fully-relayed streaming body must be closed"); + } + + @Test + void fixedLengthStreamingBody_isClosed_evenWhenTheClientDisconnectsMidTransfer() { + // Regression test: a handler's streaming body (e.g. a reverse proxy relaying a pooled + // upstream connection's response) must have close() called even when the downstream + // write fails partway through — otherwise a resource that's only released from close(), + // not from observing EOF on a read() the failed write means it never reaches, leaks. + TrackingInputStream stream = new TrackingInputStream("hello world".getBytes(StandardCharsets.UTF_8)); + Response response = new Response(200, ContentType.TEXT_PLAIN).stream(stream, 11); + FailingOutputStream out = new FailingOutputStream(1); // 1st call writes the head, 2nd (body) fails + + assertThrows(IOException.class, () -> + Http1ResponseWriter.writeResponse(out, response, HttpMethod.GET, true, false, scratch())); + assertTrue(stream.closed, "the streaming body must be closed even when the write to the client fails"); + } + + @Test + void chunkedStreamingBody_isClosed_evenWhenTheClientDisconnectsMidTransfer() { + TrackingInputStream stream = new TrackingInputStream("hello world".getBytes(StandardCharsets.UTF_8)); + Response response = new Response(200, ContentType.TEXT_PLAIN).chunked(stream); + FailingOutputStream out = new FailingOutputStream(1); // 1st call writes the head, 2nd (body) fails + + assertThrows(IOException.class, () -> + Http1ResponseWriter.writeResponse(out, response, HttpMethod.GET, true, false, scratch())); + assertTrue(stream.closed, "the streaming body must be closed even when the write to the client fails"); + } } diff --git a/flash/src/test/java/dev/relism/flash/http2/H2cPriorKnowledgeTest.java b/flash/src/test/java/dev/relism/flash/http2/H2cPriorKnowledgeTest.java index 1a9e874..fdd6186 100644 --- a/flash/src/test/java/dev/relism/flash/http2/H2cPriorKnowledgeTest.java +++ b/flash/src/test/java/dev/relism/flash/http2/H2cPriorKnowledgeTest.java @@ -3,15 +3,18 @@ package dev.relism.flash.http2; import static org.junit.jupiter.api.Assertions.assertArrayEquals; import static org.junit.jupiter.api.Assertions.assertEquals; +import dev.relism.flash.bytes.ByteWriter; import dev.relism.flash.extension.FlashApp; import dev.relism.flash.extension.FlashConfiguration; -import dev.relism.flash.http2.client.Http2Client; -import dev.relism.flash.http2.client.Http2ClientResponse; +import dev.relism.flash.http2.frame.FrameFlags; +import dev.relism.flash.http2.frame.FrameType; +import dev.relism.flash.http2.hpack.HpackDecoder; +import dev.relism.flash.http2.hpack.HpackEncoder; import java.io.InputStream; import java.net.ServerSocket; import java.net.Socket; -import java.net.URI; import java.nio.charset.StandardCharsets; +import java.util.Arrays; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.Test; @@ -55,14 +58,73 @@ class H2cPriorKnowledgeTest { app.get("/", (request, response) -> "h2c"); app.start(); - try (Http2Client client = new Http2Client()) { - Http2ClientResponse response = - client.get(URI.create("http://127.0.0.1:" + enabledPort + "/")); - assertEquals(200, response.statusCode()); - assertEquals("h2c", new String(response.body(), StandardCharsets.UTF_8)); + try (Socket socket = new Socket("127.0.0.1", enabledPort)) { + socket.setSoTimeout(5_000); + ByteWriter block = new ByteWriter(32); + HpackEncoder.writeIndexed(block, 2); // :method GET + HpackEncoder.writeIndexed(block, 6); // :scheme http + HpackEncoder.writeIndexed(block, 4); // :path / + HpackEncoder.writeLiteralWithNameIndex( + block, 1, ("127.0.0.1:" + enabledPort).getBytes(StandardCharsets.US_ASCII), false); + socket + .getOutputStream() + .write( + Http2TestFrames.concat( + Http2TestFrames.PREFACE, + Http2TestFrames.settings(), + Http2TestFrames.frame( + FrameType.HEADERS, + FrameFlags.END_HEADERS | FrameFlags.END_STREAM, + 1, + Arrays.copyOf(block.array(), block.length())))); + socket.getOutputStream().flush(); + + assertEquals(200, readStatus(socket.getInputStream())); + assertEquals("h2c", new String(readData(socket.getInputStream()), StandardCharsets.UTF_8)); } } + private static int readStatus(InputStream input) throws Exception { + HpackDecoder decoder = new HpackDecoder(); + while (true) { + Http2TestFrames.WireFrame frame = readFrame(input); + if (frame.type() != FrameType.HEADERS.code() || frame.streamId() != 1) continue; + int[] status = {0}; + decoder.decode( + frame.payload(), + 0, + frame.payload().length, + (name, value, never) -> { + if (name.length() == 7 && name.byteAt(0) == ':') { + status[0] = + (value.byteAt(0) - '0') * 100 + + (value.byteAt(1) - '0') * 10 + + value.byteAt(2) + - '0'; + } + }); + return status[0]; + } + } + + private static byte[] readData(InputStream input) throws Exception { + for (int i = 0; i < 12; i++) { + Http2TestFrames.WireFrame frame = readFrame(input); + if (frame.streamId() == 1 && frame.type() == FrameType.DATA.code() + && frame.payload().length != 0) return frame.payload(); + } + throw new AssertionError("missing h2c response DATA"); + } + + private static Http2TestFrames.WireFrame readFrame(InputStream input) throws Exception { + byte[] header = input.readNBytes(9); + int length = ((header[0] & 0xff) << 16) | ((header[1] & 0xff) << 8) | (header[2] & 0xff); + byte[] payload = input.readNBytes(length); + return new Http2TestFrames.WireFrame( + header[3] & 0xff, header[4] & 0xff, Http2TestFrames.readInt(header, 5) & 0x7fff_ffff, + payload); + } + private static int freePort() throws Exception { try (ServerSocket socket = new ServerSocket(0)) { return socket.getLocalPort(); diff --git a/flash/src/test/java/dev/relism/flash/transport/ConnectionRunnerTest.java b/flash/src/test/java/dev/relism/flash/transport/ConnectionRunnerTest.java index da1da49..e38f71c 100644 --- a/flash/src/test/java/dev/relism/flash/transport/ConnectionRunnerTest.java +++ b/flash/src/test/java/dev/relism/flash/transport/ConnectionRunnerTest.java @@ -16,6 +16,7 @@ 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.junit.jupiter.api.Test; /** @@ -79,4 +80,61 @@ class ConnectionRunnerTest { executor.shutdownNow(); } } + + @Test + void connectionsBeyondMaxConnections_areClosedImmediately_beforeAnyProtocolWork() + throws Exception { + ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor(); + Set activeSockets = ConcurrentHashMap.newKeySet(); + ScratchPool scratchPool = new ScratchPool(); + AbstractRouter router = new FastPathRouterImpl(); + AbstractWsRouter wsRouter = new FastPathWsRouterImpl(); + FlashConfiguration configuration = + FlashConfiguration.builder().port(0).maxConnections(1).build(); + + AtomicInteger protocolInvocations = new AtomicInteger(); + ConnectionProtocol countingProtocol = + ctx -> { + protocolInvocations.incrementAndGet(); + throw new IOException("simulated protocol failure"); + }; + + ConnectionRunner runner = + new ConnectionRunner( + executor, + activeSockets, + scratchPool, + router, + wsRouter, + configuration, + countingProtocol, + () -> countingProtocol); + + try (ServerSocket serverSocket = new ServerSocket(0)) { + int port = serverSocket.getLocalPort(); + + // First connection: admitted (activeSockets is empty, limit is 1). Held open by never + // closing the client socket, so it still counts toward the limit for the second attempt. + Socket firstClient = new Socket("127.0.0.1", port); + Socket firstServerSide = serverSocket.accept(); + activeSockets.add(firstServerSide); // simulate an in-flight, still-admitted connection + + // Second connection: activeSockets.size() (1) >= maxConnections (1) -> must be + // rejected at accept() time, before the executor or protocol ever run. + try (Socket secondClient = new Socket("127.0.0.1", port); + Socket secondServerSide = serverSocket.accept()) { + runner.accept(secondServerSide, () -> false); + Thread.sleep(300); // give any (incorrectly) submitted virtual-thread task time to run + + assertEquals(0, protocolInvocations.get(), "rejected connection must not reach the protocol"); + assertEquals(1, activeSockets.size(), "rejected connection must not be added to activeSockets"); + assertTrue(secondServerSide.isClosed(), "rejected connection's socket must be closed"); + } finally { + firstServerSide.close(); + firstClient.close(); + } + } finally { + executor.shutdownNow(); + } + } }