From 819baa52459d1cc59ed825a45f6f0b2bb41206c4 Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Mon, 17 Aug 2026 10:32:45 +0000 Subject: [PATCH 1/2] fix(data): make BlockingPipedOutputStream.close() idempotent The check of the `closed` flag and the closing handshake were not atomic, so two threads closing the same stream both put the end-of-stream marker into the queue. Once the reader stopped consuming, the second one blocked on the full queue and failed with "Close stream timed out after ms". Claim the close with an AtomicBoolean so exactly one caller performs the handshake. Fixes: https://github.com/ClickHouse/clickhouse-java/issues/3055 --- CHANGELOG.md | 6 +++ .../stream/BlockingPipedOutputStream.java | 4 +- .../stream/BlockingPipedOutputStreamTest.java | 49 +++++++++++++++++++ 3 files changed, 58 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 52c92a57a..ff5c5b7a6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,12 @@ ### Bug Fixes +- **[data]** Fixed `BlockingPipedOutputStream.close()` not being idempotent under concurrency: the check of the + `closed` flag and the closing handshake were not atomic, so two threads closing the same stream (e.g. a writer + thread and a try-with-resources block) could both put the end-of-stream marker into the queue, and the second one + failed with `Close stream timed out after ms` once the reader had stopped consuming. Exactly one caller now + performs the handshake and runs the post-close action; a concurrent or repeated `close()` returns immediately. + (https://github.com/ClickHouse/clickhouse-java/issues/3055) - **[client-v2]** Fixed LZ4 input streams not closing their underlying HTTP response stream. Closing an LZ4 stream returned by `QueryResponse.getInputStream()` now releases the wrapped transport stream, including after a partial read. (https://github.com/ClickHouse/clickhouse-java/issues/2985) diff --git a/clickhouse-data/src/main/java/com/clickhouse/data/stream/BlockingPipedOutputStream.java b/clickhouse-data/src/main/java/com/clickhouse/data/stream/BlockingPipedOutputStream.java index ea024f5d5..169f4ad79 100644 --- a/clickhouse-data/src/main/java/com/clickhouse/data/stream/BlockingPipedOutputStream.java +++ b/clickhouse-data/src/main/java/com/clickhouse/data/stream/BlockingPipedOutputStream.java @@ -8,6 +8,7 @@ import java.util.concurrent.CompletableFuture; import java.util.concurrent.LinkedBlockingQueue; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import com.clickhouse.data.ClickHouseByteBuffer; import com.clickhouse.data.ClickHouseChecker; @@ -33,6 +34,7 @@ public class BlockingPipedOutputStream extends ClickHousePipedOutputStream { private final int bufferSize; private final CompletableFuture future; private final long timeout; + private final AtomicBoolean closing = new AtomicBoolean(false); private ByteBuffer buffer; @@ -105,7 +107,7 @@ public ClickHouseInputStream getInputStream(Runnable postCloseAction) { @Override public void close() throws IOException { - if (closed) { + if (closed || !closing.compareAndSet(false, true)) { return; } diff --git a/clickhouse-data/src/test/java/com/clickhouse/data/stream/BlockingPipedOutputStreamTest.java b/clickhouse-data/src/test/java/com/clickhouse/data/stream/BlockingPipedOutputStreamTest.java index 9c84c85e0..d210bc82c 100644 --- a/clickhouse-data/src/test/java/com/clickhouse/data/stream/BlockingPipedOutputStreamTest.java +++ b/clickhouse-data/src/test/java/com/clickhouse/data/stream/BlockingPipedOutputStreamTest.java @@ -6,10 +6,16 @@ import java.io.UncheckedIOException; import java.nio.ByteBuffer; import java.nio.Buffer; +import java.util.ArrayList; import java.util.Arrays; +import java.util.Collection; +import java.util.List; +import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -209,6 +215,49 @@ public void testWriteBytes() throws InterruptedException, IOException { } } + @Test(groups = { "unit" }) + public void testConcurrentClose() throws Exception { + final int closers = 4; + final long timeout = 500L; + final AtomicInteger closeCount = new AtomicInteger(0); + final Collection errors = new ConcurrentLinkedQueue<>(); + final BlockingPipedOutputStream stream = new BlockingPipedOutputStream(4, 1, timeout, + (Runnable) closeCount::incrementAndGet); + // fill the only slot of the queue so that the closing handshake cannot complete + stream.queue.put(ByteBuffer.allocate(1)); + + final CyclicBarrier barrier = new CyclicBarrier(closers); + final ExecutorService executor = Executors.newFixedThreadPool(closers); + try { + List> futures = new ArrayList<>(closers); + for (int i = 0; i < closers; i++) { + futures.add(executor.submit(() -> { + barrier.await(); + try { + stream.close(); + } catch (IOException e) { + errors.add(String.valueOf(e.getMessage())); + } + return null; + })); + } + for (Future f : futures) { + f.get(timeout + 30000L, TimeUnit.MILLISECONDS); + } + } finally { + executor.shutdownNow(); + } + + Assert.assertEquals(closeCount.get(), 1, "Stream should have been closed exactly once"); + Assert.assertEquals(errors.size(), 1, "Only the thread which closed the stream may fail"); + Assert.assertTrue(errors.iterator().next().indexOf("Close stream timed out") == 0, + "Unexpected error: " + errors); + Assert.assertEquals(stream.queue.size(), 1, "No additional buffer should have been queued"); + + stream.close(); + Assert.assertEquals(closeCount.get(), 1, "Closing a closed stream should do nothing"); + } + @Test(groups = { "unit" }) public void testPipedStream() throws InterruptedException, IOException { final int timeout = 10000; From 7c4b7a661c4009d3531d5d8aae1a3b3892c46b89 Mon Sep 17 00:00:00 2001 From: Polyglot AI <293096396+polyglotAI-bot@users.noreply.github.com> Date: Mon, 17 Aug 2026 10:56:27 +0000 Subject: [PATCH 2/2] fix(data): keep close() bookkeeping atomic when flushing the remaining buffer fails Flushing the pending buffer sat outside the try/finally which sets closed and runs the post close action. With the new compareAndSet claim a failing flush left the claim taken while the stream was still open, so every later close() returned immediately and the stream could never be closed. Move the flush inside the try block, matching ClickHouseOutputStream.close() and NonBlockingPipedOutputStream.close(). Fixes: https://github.com/ClickHouse/clickhouse-java/issues/3055 --- CHANGELOG.md | 5 ++-- .../stream/BlockingPipedOutputStream.java | 8 +++--- .../stream/BlockingPipedOutputStreamTest.java | 25 +++++++++++++++++++ 3 files changed, 32 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ff5c5b7a6..7f7edf4ab 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,8 +48,9 @@ `closed` flag and the closing handshake were not atomic, so two threads closing the same stream (e.g. a writer thread and a try-with-resources block) could both put the end-of-stream marker into the queue, and the second one failed with `Close stream timed out after ms` once the reader had stopped consuming. Exactly one caller now - performs the handshake and runs the post-close action; a concurrent or repeated `close()` returns immediately. - (https://github.com/ClickHouse/clickhouse-java/issues/3055) + performs the handshake and runs the post-close action; a concurrent or repeated `close()` returns immediately. A + `close()` which fails while flushing the remaining data also marks the stream closed and runs the post-close + action, so the stream cannot stay half-closed. (https://github.com/ClickHouse/clickhouse-java/issues/3055) - **[client-v2]** Fixed LZ4 input streams not closing their underlying HTTP response stream. Closing an LZ4 stream returned by `QueryResponse.getInputStream()` now releases the wrapped transport stream, including after a partial read. (https://github.com/ClickHouse/clickhouse-java/issues/2985) diff --git a/clickhouse-data/src/main/java/com/clickhouse/data/stream/BlockingPipedOutputStream.java b/clickhouse-data/src/main/java/com/clickhouse/data/stream/BlockingPipedOutputStream.java index 169f4ad79..fdbadc6df 100644 --- a/clickhouse-data/src/main/java/com/clickhouse/data/stream/BlockingPipedOutputStream.java +++ b/clickhouse-data/src/main/java/com/clickhouse/data/stream/BlockingPipedOutputStream.java @@ -111,12 +111,12 @@ public void close() throws IOException { return; } - if (buffer.position() > 0) { - updateBuffer(false); - } - // buffer = ClickHouseByteBuffer.EMPTY_BUFFER; try { + if (buffer.position() > 0) { + updateBuffer(false); + } + if (timeout > 0L) { if (!queue.offer(ClickHouseByteBuffer.EMPTY_BUFFER, timeout, TimeUnit.MILLISECONDS)) { throw new IOException(ClickHouseUtils.format("Close stream timed out after %d ms", timeout)); diff --git a/clickhouse-data/src/test/java/com/clickhouse/data/stream/BlockingPipedOutputStreamTest.java b/clickhouse-data/src/test/java/com/clickhouse/data/stream/BlockingPipedOutputStreamTest.java index d210bc82c..f3fda67b9 100644 --- a/clickhouse-data/src/test/java/com/clickhouse/data/stream/BlockingPipedOutputStreamTest.java +++ b/clickhouse-data/src/test/java/com/clickhouse/data/stream/BlockingPipedOutputStreamTest.java @@ -258,6 +258,31 @@ public void testConcurrentClose() throws Exception { Assert.assertEquals(closeCount.get(), 1, "Closing a closed stream should do nothing"); } + @Test(groups = { "unit" }) + public void testCloseWhenFlushingRemainingBufferFails() throws Exception { + final long timeout = 500L; + final AtomicInteger closeCount = new AtomicInteger(0); + final BlockingPipedOutputStream stream = new BlockingPipedOutputStream(4, 1, timeout, + (Runnable) closeCount::incrementAndGet); + stream.writeByte((byte) 1); + // fill the only slot of the queue so that flushing the remaining buffer fails + stream.queue.put(ByteBuffer.allocate(1)); + + try { + stream.close(); + Assert.fail("Close should fail"); + } catch (IOException e) { + Assert.assertTrue(e.getMessage().indexOf("Write timed out") == 0, "Unexpected error: " + e.getMessage()); + } + + Assert.assertTrue(stream.isClosed(), "Stream should have been closed"); + Assert.assertEquals(closeCount.get(), 1, "Post close action should have been executed exactly once"); + Assert.assertEquals(stream.queue.size(), 1, "No additional buffer should have been queued"); + + stream.close(); + Assert.assertEquals(closeCount.get(), 1, "Closing a closed stream should do nothing"); + } + @Test(groups = { "unit" }) public void testPipedStream() throws InterruptedException, IOException { final int timeout = 10000;