This is an automated email from the ASF dual-hosted git repository.
amaliujia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ozone.git
The following commit(s) were added to refs/heads/master by this push:
new 3c2a71d9d99 HDDS-16789. Make Buffers.pollAll() allows empty buffer
instead of failing on empty deque (#11446)
3c2a71d9d99 is described below
commit 3c2a71d9d99c187c7eb741c110ceb0f05abfd2aa
Author: Rui Wang <[email protected]>
AuthorDate: Thu Oct 15 10:02:57 2026 +0800
HDDS-16789. Make Buffers.pollAll() allows empty buffer instead of failing
on empty deque (#11446)
---
.../ozone/container/keyvalue/impl/Buffers.java | 14 +++++++--
.../keyvalue/impl/KeyValueStreamDataChannel.java | 8 ++---
.../impl/TestKeyValueStreamDataChannel.java | 34 +++++++++++++++++++++-
3 files changed, 49 insertions(+), 7 deletions(-)
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/Buffers.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/Buffers.java
index 9a65f96d393..6756393b88c 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/Buffers.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/Buffers.java
@@ -89,8 +89,18 @@ ReferenceCountedObject<ByteBuffer> poll() {
return polled;
}
- ReferenceCountedObject<ByteBuf> pollAll() {
- Preconditions.checkState(!deque.isEmpty(), "The deque is empty");
+ /**
+ * Poll all the buffers from the deque and combine them into a single
read-only {@link ByteBuf}.
+ *
+ * @param allowEmpty whether an empty deque is allowed.
+ * @return the combined buffer; or null if the deque is empty and allowEmpty
is true.
+ * @throws IllegalStateException if the deque is empty and allowEmpty is
false.
+ */
+ ReferenceCountedObject<ByteBuf> pollAll(boolean allowEmpty) {
+ if (deque.isEmpty()) {
+ Preconditions.checkState(allowEmpty, "The deque is empty");
+ return null;
+ }
final ByteBuffer[] array = new ByteBuffer[deque.size()];
final List<ReferenceCountedObject<ByteBuffer>> refs = new
ArrayList<>(deque.size());
for (int i = 0; i < array.length; i++) {
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java
index 190c0bd164c..3ff82534f09 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/impl/KeyValueStreamDataChannel.java
@@ -141,7 +141,7 @@ protected void cleanupInternal() throws IOException {
* Note that the PutBlock proto at the end is ignored; see HDDS-12007.
*/
private void writeBuffers() throws IOException {
- final ReferenceCountedObject<ByteBuf> ref = buffers.pollAll();
+ final ReferenceCountedObject<ByteBuf> ref = buffers.pollAll(false);
final ByteBuf buf = ref.retain();
try {
setEndIndex(buf);
@@ -159,10 +159,10 @@ private void writeBuffers() throws IOException {
*/
public void drainBuffers() throws IOException {
assertOpen();
- if (buffers.isEmpty()) {
+ final ReferenceCountedObject<ByteBuf> ref = buffers.pollAll(true);
+ if (ref == null) {
return;
}
- final ReferenceCountedObject<ByteBuf> ref = buffers.pollAll();
final ByteBuf buf = ref.retain();
try {
writeFully(buf.nioBuffer(), super::writeFileChannel);
@@ -173,7 +173,7 @@ public void drainBuffers() throws IOException {
static ContainerCommandRequestProto closeBuffers(
Buffers buffers, WriteMethod writeMethod) throws IOException {
- final ReferenceCountedObject<ByteBuf> ref = buffers.pollAll();
+ final ReferenceCountedObject<ByteBuf> ref = buffers.pollAll(false);
final ByteBuf buf = ref.retain();
final ContainerCommandRequestProto putBlockRequestProto;
try {
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestKeyValueStreamDataChannel.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestKeyValueStreamDataChannel.java
index 5da8acec08e..e873cd433f0 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestKeyValueStreamDataChannel.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/keyvalue/impl/TestKeyValueStreamDataChannel.java
@@ -26,6 +26,7 @@
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.mock;
@@ -165,6 +166,37 @@ public void testDrainBuffers() throws Exception {
assertArrayEquals(expected, Files.readAllBytes(tempFile.toPath()));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testPollAll(boolean allowEmpty) {
+ final Buffers buffers = new Buffers(PUT_BLOCK_REQUEST_LENGTH_MAX);
+ final byte[] data = RandomUtils.secure().randomBytes(50);
+ final ReferenceCountedObject<ByteBuffer> offered =
+ ReferenceCountedObject.wrap(ByteBuffer.wrap(data), () -> { }, () -> {
});
+ offered.retain();
+ assertFalse(buffers.offer(offered).iterator().hasNext());
+ offered.release();
+ assertFalse(buffers.isEmpty());
+
+ final ReferenceCountedObject<ByteBuf> ref = buffers.pollAll(allowEmpty);
+ final ByteBuf buf = ref.retain();
+ try {
+ final byte[] polled = new byte[buf.readableBytes()];
+ buf.readBytes(polled);
+ assertArrayEquals(data, polled);
+ } finally {
+ ref.release();
+ }
+ assertTrue(buffers.isEmpty());
+
+ if (allowEmpty) {
+ assertNull(buffers.pollAll(true));
+ } else {
+ assertThrows(IllegalStateException.class, () -> buffers.pollAll(false));
+ }
+ assertTrue(buffers.isEmpty());
+ }
+
@Test
public void testReadPutBlockRequestBufferTooShort() {
final ByteBuf buf = Unpooled.buffer(2);
@@ -351,7 +383,7 @@ public CompletableFuture<DataStreamReply> closeAsync() {
static ContainerCommandRequestProto closeBuffers(
Buffers buffers, WriteMethod writeMethod) throws IOException {
- final ReferenceCountedObject<ByteBuf> ref = buffers.pollAll();
+ final ReferenceCountedObject<ByteBuf> ref = buffers.pollAll(false);
final ByteBuf buf = ref.retain();
final ContainerCommandRequestProto putBlockRequest;
try {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
