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 3a22acfaa21 HDDS-16746. Local Stream onCommand should drain buffers
before commit PutBlock (#11427)
3a22acfaa21 is described below
commit 3a22acfaa214ed7fcdd25ba193bf7088e8e2bd87
Author: Rui Wang <[email protected]>
AuthorDate: Thu Oct 8 21:35:55 2026 +0800
HDDS-16746. Local Stream onCommand should drain buffers before commit
PutBlock (#11427)
---
.../common/transport/server/ratis/LocalStream.java | 6 +-
.../ozone/container/keyvalue/impl/Buffers.java | 5 ++
.../keyvalue/impl/KeyValueStreamDataChannel.java | 19 ++++++
.../server/ratis/ContainerStateMachineTests.java | 78 ++++++++++++++++++++++
.../impl/TestKeyValueStreamDataChannel.java | 34 ++++++++++
.../rpc/TestContainerStateMachineStream.java | 23 +++++++
6 files changed, 164 insertions(+), 1 deletion(-)
diff --git
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/LocalStream.java
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/LocalStream.java
index 47189031a11..b2817e8bd3f 100644
---
a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/LocalStream.java
+++
b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/LocalStream.java
@@ -58,9 +58,13 @@ public CompletableFuture<?> cleanUp() {
@Override
public CompletableFuture<ByteBuffer> onCommand(ByteBuffer buffer, long
streamOffset) {
+ if (!(dataChannel instanceof KeyValueStreamDataChannel)) {
+ return JavaUtils.completeExceptionally(new IllegalStateException(
+ "Unexpected DataChannel " + dataChannel.getClass()));
+ }
return CompletableFuture.supplyAsync(() -> {
try {
- // TODO: drain datachannel.buffers
+ ((KeyValueStreamDataChannel) dataChannel).drainBuffers();
return command.apply(buffer);
} catch (IOException e) {
throw new CompletionException(e);
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 308c8bb13a0..9a65f96d393 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
@@ -98,6 +98,7 @@ ReferenceCountedObject<ByteBuf> pollAll() {
refs.add(ref);
array[i] = ref.get();
}
+ length = 0;
final ByteBuf buf = Unpooled.wrappedBuffer(array).asReadOnly();
return ReferenceCountedObject.wrap(buf, () -> { }, () -> {
buf.release();
@@ -105,6 +106,10 @@ ReferenceCountedObject<ByteBuf> pollAll() {
});
}
+ boolean isEmpty() {
+ return deque.isEmpty();
+ }
+
void cleanUpAll() {
while (!deque.isEmpty()) {
poll().release();
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 5a735812d79..190c0bd164c 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
@@ -152,6 +152,25 @@ private void writeBuffers() throws IOException {
}
}
+ /**
+ * Write all the data in {@link #buffers} to the channel.
+ * It must be called before committing a PutBlock in the middle of the
stream;
+ * otherwise, the buffered data is not yet in the block file.
+ */
+ public void drainBuffers() throws IOException {
+ assertOpen();
+ if (buffers.isEmpty()) {
+ return;
+ }
+ final ReferenceCountedObject<ByteBuf> ref = buffers.pollAll();
+ final ByteBuf buf = ref.retain();
+ try {
+ writeFully(buf.nioBuffer(), super::writeFileChannel);
+ } finally {
+ ref.release();
+ }
+ }
+
static ContainerCommandRequestProto closeBuffers(
Buffers buffers, WriteMethod writeMethod) throws IOException {
final ReferenceCountedObject<ByteBuf> ref = buffers.pollAll();
diff --git
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachineTests.java
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachineTests.java
index 8ad92ec358e..57cbae9b3c4 100644
---
a/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachineTests.java
+++
b/hadoop-hdds/container-service/src/test/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachineTests.java
@@ -22,9 +22,14 @@
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.argThat;
import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.reset;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
@@ -32,6 +37,7 @@
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.io.IOException;
+import java.nio.ByteBuffer;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
@@ -45,17 +51,21 @@
import java.util.stream.IntStream;
import org.apache.hadoop.hdds.conf.OzoneConfiguration;
import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos;
+import org.apache.hadoop.hdds.ratis.ContainerCommandRequestMessage;
import
org.apache.hadoop.hdds.scm.container.common.helpers.StorageContainerException;
import org.apache.hadoop.ozone.container.common.interfaces.ContainerDispatcher;
+import
org.apache.hadoop.ozone.container.keyvalue.impl.KeyValueStreamDataChannel;
import org.apache.hadoop.ozone.container.ozoneimpl.ContainerController;
import org.apache.ozone.test.tag.Flaky;
import org.apache.ratis.proto.RaftProtos;
import org.apache.ratis.protocol.Message;
+import org.apache.ratis.protocol.RaftClientRequest;
import org.apache.ratis.protocol.RaftGroup;
import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.protocol.RaftPeer;
import org.apache.ratis.server.DivisionInfo;
import org.apache.ratis.server.RaftServer;
+import org.apache.ratis.statemachine.StateMachine;
import org.apache.ratis.statemachine.TransactionContext;
import org.apache.ratis.thirdparty.com.google.protobuf.ByteString;
import org.junit.jupiter.api.AfterAll;
@@ -65,6 +75,7 @@
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.InOrder;
/**
* Test class to ContainerStateMachine class.
@@ -247,6 +258,73 @@ public void testWriteTimout() throws Exception {
assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR,
sce.getResult());
}
+ @Test
+ public void testStreamCommandDrainsBuffersBeforePutBlock() throws Exception {
+ final KeyValueStreamDataChannel channel =
mock(KeyValueStreamDataChannel.class);
+ final StateMachine.DataStream stream = setUpStream(channel);
+
+ stream.onCommand(newPutBlockCommand(), 0).get();
+
+ final InOrder inOrder = inOrder(channel, dispatcher);
+ inOrder.verify(channel).drainBuffers();
+ inOrder.verify(dispatcher).dispatch(argThat(r -> r.getCmdType() ==
ContainerProtos.Type.PutBlock),
+ argThat(c -> DispatcherContext.op(c) ==
DispatcherContext.Op.STREAM_COMMAND));
+ }
+
+ @Test
+ public void testStreamCommandNotDispatchedIfDrainFails() throws Exception {
+ final KeyValueStreamDataChannel channel =
mock(KeyValueStreamDataChannel.class);
+ doThrow(new IOException("drain failed")).when(channel).drainBuffers();
+ final StateMachine.DataStream stream = setUpStream(channel);
+
+ final ExecutionException e = assertThrows(ExecutionException.class,
+ () -> stream.onCommand(newPutBlockCommand(), 0).get());
+ assertInstanceOf(IOException.class, e.getCause());
+ verify(dispatcher, never()).dispatch(argThat(r -> r.getCmdType() ==
ContainerProtos.Type.PutBlock), any());
+ }
+
+ @Test
+ public void testStreamCommandWithUnexpectedDataChannel() throws Exception {
+ final StateMachine.DataStream stream =
setUpStream(mock(StateMachine.DataChannel.class));
+
+ final ExecutionException e = assertThrows(ExecutionException.class,
+ () -> stream.onCommand(newPutBlockCommand(), 0).get());
+ assertInstanceOf(IllegalStateException.class, e.getCause());
+ verify(dispatcher, never()).dispatch(argThat(r -> r.getCmdType() ==
ContainerProtos.Type.PutBlock), any());
+ }
+
+ private StateMachine.DataStream setUpStream(StateMachine.DataChannel
channel) throws Exception {
+ when(dispatcher.dispatch(any(),
any())).thenReturn(ContainerProtos.ContainerCommandResponseProto.newBuilder()
+
.setCmdType(ContainerProtos.Type.PutBlock).setResult(ContainerProtos.Result.SUCCESS).build());
+ when(dispatcher.getStreamDataChannel(any(), any())).thenReturn(channel);
+ final ContainerProtos.ContainerCommandRequestProto streamInit =
ContainerProtos.ContainerCommandRequestProto
+ .newBuilder()
+ .setCmdType(ContainerProtos.Type.StreamInit)
+
.setWriteChunk(ContainerProtos.WriteChunkRequestProto.newBuilder().setBlockID(newBlockID()).build())
+ .setContainerID(1)
+ .setDatanodeUuid(UUID.randomUUID().toString())
+ .build();
+ final RaftClientRequest request = mock(RaftClientRequest.class);
+
when(request.getMessage()).thenReturn(ContainerCommandRequestMessage.toMessage(streamInit,
null));
+ return stateMachine.stream(request).get();
+ }
+
+ private static ByteBuffer newPutBlockCommand() {
+ final ContainerProtos.ContainerCommandRequestProto putBlock =
ContainerProtos.ContainerCommandRequestProto
+ .newBuilder()
+ .setCmdType(ContainerProtos.Type.PutBlock)
+ .setPutBlock(ContainerProtos.PutBlockRequestProto.newBuilder()
+
.setBlockData(ContainerProtos.BlockData.newBuilder().setBlockID(newBlockID()).build()))
+ .setContainerID(1)
+ .setDatanodeUuid(UUID.randomUUID().toString())
+ .build();
+ return ContainerCommandRequestMessage.toMessage(putBlock,
null).getContent().asReadOnlyByteBuffer();
+ }
+
+ private static ContainerProtos.DatanodeBlockID newBlockID() {
+ return
ContainerProtos.DatanodeBlockID.newBuilder().setContainerID(1).setLocalID(1).build();
+ }
+
private void setUpMockDispatcherReturn(boolean failWithException) {
if (failWithException) {
when(dispatcher.dispatch(any(), any())).thenThrow(new
RuntimeException());
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 638c5707725..59ce314d632 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
@@ -131,6 +131,40 @@ public void testClosePutBlockBehavior(boolean
commitPutBlockOnClose) throws Exce
assertArrayEquals(data, Files.readAllBytes(tempFile.toPath()));
}
+ @Test
+ public void testDrainBuffers() throws Exception {
+ File tempFile = File.createTempFile("test-kv-drain", ".tmp");
+ tempFile.deleteOnExit();
+ AtomicReference<ContainerCommandRequestProto> processed = new
AtomicReference<>();
+ KeyValueStreamDataChannel channel = newChannel(tempFile, true, processed);
+ final byte[] data1 = RandomUtils.secure().randomBytes(50);
+ final byte[] data2 = RandomUtils.secure().randomBytes(30);
+ final ByteBuffer putBlockBuf = ContainerCommandRequestMessage.toMessage(
+ PUT_BLOCK_PROTO, null).getContent().asReadOnlyByteBuffer();
+ final ByteBuffer protoLengthBuf = getProtoLength(putBlockBuf,
PUT_BLOCK_REQUEST_LENGTH_MAX);
+
+ assertEquals(0, tempFile.length());
+ write(channel, data1);
+ // data1 is smaller than PUT_BLOCK_REQUEST_LENGTH_MAX, so it is still
buffered
+ assertEquals(0, tempFile.length());
+ channel.drainBuffers();
+ assertEquals(data1.length, tempFile.length());
+ // draining empty buffers is a no-op
+ channel.drainBuffers();
+ assertEquals(data1.length, tempFile.length());
+
+ write(channel, data2);
+ write(channel, putBlockBuf.duplicate());
+ write(channel, protoLengthBuf.duplicate());
+ channel.close();
+
+ assertEquals(PUT_BLOCK_PROTO, processed.get());
+ final byte[] expected = new byte[data1.length + data2.length];
+ System.arraycopy(data1, 0, expected, 0, data1.length);
+ System.arraycopy(data2, 0, expected, data1.length, data2.length);
+ assertArrayEquals(expected, Files.readAllBytes(tempFile.toPath()));
+ }
+
@Test
public void testReadPutBlockRequestBufferTooShort() {
final ByteBuf buf = Unpooled.buffer(2);
diff --git
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachineStream.java
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachineStream.java
index c10dfe784fb..060ef7ce8c7 100644
---
a/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachineStream.java
+++
b/hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestContainerStateMachineStream.java
@@ -20,6 +20,7 @@
import static
org.apache.hadoop.hdds.scm.ScmConfigKeys.OZONE_SCM_CHUNK_SIZE_KEY;
import static
org.apache.hadoop.ozone.container.OzoneTestHelper.createStreamKey;
import static
org.apache.hadoop.ozone.container.OzoneTestHelper.getDatanodeService;
+import static org.apache.hadoop.ozone.container.OzoneTestHelper.validateData;
import static org.assertj.core.api.Assertions.assertThat;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -42,6 +43,7 @@
import org.apache.ozone.test.NonHATests;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.Arguments;
@@ -119,4 +121,25 @@ void testContainerStateMachineForStreaming(int offset,
boolean putBlockWithoutRa
.isGreaterThanOrEqualTo(size);
}
+ @Test
+ void testReadAfterHsyncWithPutBlockWithoutRaft() throws Exception {
+ // Smaller than PUT_BLOCK_REQUEST_LENGTH_MAX, so the datanodes still
buffer all the data when hsync commits PutBlock
+ final int size = 1000;
+ OzoneConfiguration conf = new OzoneConfiguration(cluster().getConf());
+ OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class);
+ clientConfig.setDatastreamPutBlockWithoutRaftEnabled(true);
+ conf.setFromObject(clientConfig);
+
+ final String keyName = "hsync-" + UUID.randomUUID();
+ try (OzoneClient streamingClient = OzoneClientFactory.getRpcClient(conf);
+ OzoneDataStreamOutput key = createStreamKey(keyName,
ReplicationType.RATIS, size,
+ streamingClient.getObjectStore(), volumeName, bucketName)) {
+ final byte[] data = ContainerTestHelper.generateData(size, true);
+ key.write(ByteBuffer.wrap(data));
+ assertInstanceOf(KeyDataStreamOutput.class,
key.getByteBufStreamOutput()).hsync();
+
+ validateData(keyName, data, client.getObjectStore(), volumeName,
bucketName);
+ }
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
