This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 29da0faa9ca fix: release buffered response bytes in
AppendableByteArrayInputStream (#20521)
29da0faa9ca is described below
commit 29da0faa9ca0cca3f18bd81f1fc0f97744a90f02
Author: Zoltan Haindrich <[email protected]>
AuthorDate: Wed Oct 14 08:04:45 2026 +0200
fix: release buffered response bytes in AppendableByteArrayInputStream
(#20521)
---
.../client/io/AppendableByteArrayInputStream.java | 27 ++++++++++--
.../io/AppendableByteArrayInputStreamTest.java | 49 ++++++++++++++++++++++
2 files changed, 72 insertions(+), 4 deletions(-)
diff --git
a/processing/src/main/java/org/apache/druid/java/util/http/client/io/AppendableByteArrayInputStream.java
b/processing/src/main/java/org/apache/druid/java/util/http/client/io/AppendableByteArrayInputStream.java
index 5c147e776df..a9ca04d8456 100644
---
a/processing/src/main/java/org/apache/druid/java/util/http/client/io/AppendableByteArrayInputStream.java
+++
b/processing/src/main/java/org/apache/druid/java/util/http/client/io/AppendableByteArrayInputStream.java
@@ -37,7 +37,7 @@ public class AppendableByteArrayInputStream extends
InputStream
private volatile boolean done = false;
private volatile Throwable throwable;
- private volatile int available = 0;
+ private volatile long available = 0;
private byte[] curr = new byte[]{};
private int currIndex = 0;
@@ -49,6 +49,9 @@ public class AppendableByteArrayInputStream extends
InputStream
}
synchronized (singleByteReaderDoer) {
+ if (done) {
+ return;
+ }
bytes.addLast(bytesToAdd);
available += bytesToAdd.length;
singleByteReaderDoer.notify();
@@ -68,10 +71,18 @@ public class AppendableByteArrayInputStream extends
InputStream
synchronized (singleByteReaderDoer) {
done = true;
throwable = t;
+ bytes.clear();
+ available = 0;
singleByteReaderDoer.notifyAll();
}
}
+ @Override
+ public void close()
+ {
+ exceptionCaught(new IOException("Stream closed"));
+ }
+
@Override
public int read() throws IOException
{
@@ -145,7 +156,7 @@ public class AppendableByteArrayInputStream extends
InputStream
break;
}
try {
- available -= numPulled;
+ releaseUnreadBytes(numPulled);
numPulled = 0;
singleByteReaderDoer.wait();
}
@@ -181,16 +192,24 @@ public class AppendableByteArrayInputStream extends
InputStream
}
synchronized (singleByteReaderDoer) {
- available -= numPulled;
+ releaseUnreadBytes(numPulled);
}
return numScanned;
}
+ private void releaseUnreadBytes(long numPulled)
+ {
+ // exceptionCaught zeroes the count, bytes pulled concurrently included.
+ if (throwable == null) {
+ available -= numPulled;
+ }
+ }
+
@Override
public int available()
{
- return available;
+ return (int) Math.min(available, Integer.MAX_VALUE);
}
private interface Doer
diff --git
a/processing/src/test/java/org/apache/druid/java/util/http/client/io/AppendableByteArrayInputStreamTest.java
b/processing/src/test/java/org/apache/druid/java/util/http/client/io/AppendableByteArrayInputStreamTest.java
index c337f7d9f70..6d4d430f382 100644
---
a/processing/src/test/java/org/apache/druid/java/util/http/client/io/AppendableByteArrayInputStreamTest.java
+++
b/processing/src/test/java/org/apache/druid/java/util/http/client/io/AppendableByteArrayInputStreamTest.java
@@ -211,6 +211,55 @@ public class AppendableByteArrayInputStreamTest
Assertions.assertEquals(-1, in.read());
}
+ @Test
+ public void testAvailableSaturatesInsteadOfOverflowing() throws Exception
+ {
+ final AppendableByteArrayInputStream in = new
AppendableByteArrayInputStream();
+
+ // The same array is queued repeatedly: the stream retains references, so
this accounts for 2 GiB of queued
+ // bytes while allocating only one chunk.
+ final byte[] oneMebibyte = new byte[1024 * 1024];
+ for (int i = 0; i < 2048; i++) {
+ in.add(oneMebibyte);
+ }
+
+ Assertions.assertEquals(Integer.MAX_VALUE, in.available());
+
+ in.read(new byte[oneMebibyte.length]);
+
+ Assertions.assertEquals((2048L - 1) * oneMebibyte.length, (long)
in.available());
+ }
+
+ @Test
+ public void testExceptionCaughtReleasesQueuedBytes() throws IOException
+ {
+ final AppendableByteArrayInputStream in = new
AppendableByteArrayInputStream();
+ in.add(new byte[10]);
+ in.add(new byte[8192]);
+ Assertions.assertEquals(5, in.read(new byte[5]));
+
+ in.exceptionCaught(new IOException("connection reset"));
+
+ Assertions.assertEquals(0, in.available());
+ Assertions.assertEquals(5, in.read(new byte[5]), "the chunk being read is
still handed out");
+ Assertions.assertEquals(0, in.available());
+ Assertions.assertThrows(IOException.class, () -> in.read(new byte[8192]));
+ }
+
+ @Test
+ public void testCloseDiscardsQueuedAndLaterChunks()
+ {
+ final AppendableByteArrayInputStream in = new
AppendableByteArrayInputStream();
+ in.add(new byte[8192]);
+ in.add(new byte[8192]);
+
+ in.close();
+ in.add(new byte[8192]);
+
+ Assertions.assertEquals(0, in.available());
+ Assertions.assertThrows(IOException.class, in::read);
+ }
+
@Test
public void testExceptionUnblocks() throws InterruptedException
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
