From 10f93437df7632a39d5e9f03d28f426fac7614eb Mon Sep 17 00:00:00 2001 From: amaliujia Date: Sat, 17 Oct 2026 20:46:11 +0800 Subject: [PATCH 1/8] Test that stream-committed PutBlock is on all datanodes of the pipeline --- .../rpc/TestContainerStateMachineStream.java | 49 +++++++++++++++++++ 1 file changed, 49 insertions(+) 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 060ef7ce8c7a..5a7be96c44e5 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,25 +20,33 @@ 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.getDatanodeServices; import static org.apache.hadoop.ozone.container.OzoneTestHelper.validateData; import static org.assertj.core.api.Assertions.assertThat; +import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertInstanceOf; import java.nio.ByteBuffer; import java.util.List; +import java.util.Set; import java.util.UUID; import java.util.stream.Stream; import org.apache.hadoop.conf.StorageUnit; import org.apache.hadoop.hdds.client.ReplicationType; import org.apache.hadoop.hdds.conf.OzoneConfiguration; +import org.apache.hadoop.hdds.protocol.datanode.proto.ContainerProtos; import org.apache.hadoop.hdds.scm.OzoneClientConfig; import org.apache.hadoop.hdds.utils.IOUtils; +import org.apache.hadoop.ozone.HddsDatanodeService; import org.apache.hadoop.ozone.client.ObjectStore; import org.apache.hadoop.ozone.client.OzoneClient; import org.apache.hadoop.ozone.client.OzoneClientFactory; import org.apache.hadoop.ozone.client.io.KeyDataStreamOutput; import org.apache.hadoop.ozone.client.io.OzoneDataStreamOutput; import org.apache.hadoop.ozone.container.ContainerTestHelper; +import org.apache.hadoop.ozone.container.common.interfaces.Container; +import org.apache.hadoop.ozone.container.keyvalue.KeyValueHandler; +import org.apache.hadoop.ozone.container.ozoneimpl.OzoneContainer; import org.apache.hadoop.ozone.om.helpers.OmKeyLocationInfo; import org.apache.ozone.test.NonHATests; import org.junit.jupiter.api.AfterAll; @@ -48,6 +56,7 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; /** * Tests the containerStateMachine stream handling. @@ -142,4 +151,44 @@ void testReadAfterHsyncWithPutBlockWithoutRaft() throws Exception { } } + @ParameterizedTest + @ValueSource(booleans = {true, false}) + void testPutBlockWithoutRaftCommittedOnAllDatanodes(boolean pipelineMode) throws Exception { + final int size = 1000; + OzoneConfiguration conf = new OzoneConfiguration(cluster().getConf()); + OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); + clientConfig.setDatastreamPutBlockWithoutRaftEnabled(true); + clientConfig.setDatastreamPipelineMode(pipelineMode); + conf.setFromObject(clientConfig); + + final String keyName = "all-dn-" + pipelineMode + "-" + UUID.randomUUID(); + final byte[] data = ContainerTestHelper.generateData(2 * size, true); + final OmKeyLocationInfo info; + try (OzoneClient streamingClient = OzoneClientFactory.getRpcClient(conf); + OzoneDataStreamOutput key = createStreamKey(keyName, ReplicationType.RATIS, 2 * size, + streamingClient.getObjectStore(), volumeName, bucketName)) { + final KeyDataStreamOutput out = assertInstanceOf(KeyDataStreamOutput.class, key.getByteBufStreamOutput()); + key.write(ByteBuffer.wrap(data, 0, size)); + out.hsync(); + info = out.getLocationInfoList().get(0); + // hsync returns only after the PutBlock is committed on every datanode of the pipeline + assertCommittedOnAllDatanodes(info, size); + + key.write(ByteBuffer.wrap(data, size, size)); + } + assertCommittedOnAllDatanodes(info, 2 * size); + validateData(keyName, data, client.getObjectStore(), volumeName, bucketName); + } + + private void assertCommittedOnAllDatanodes(OmKeyLocationInfo info, long expectedLength) throws Exception { + final Set datanodes = getDatanodeServices(cluster(), info.getPipeline()); + for (HddsDatanodeService dn : datanodes) { + final OzoneContainer ozoneContainer = dn.getDatanodeStateMachine().getContainer(); + final KeyValueHandler handler = (KeyValueHandler) ozoneContainer.getDispatcher() + .getHandler(ContainerProtos.ContainerType.KeyValueContainer); + final Container container = ozoneContainer.getContainerSet().getContainer(info.getContainerID()); + assertEquals(expectedLength, handler.getBlockManager().getCommittedBlockLength(container, info.getBlockID()), + () -> "Unexpected committed block length on " + dn.getDatanodeDetails()); + } + } } From 4daccddaa4de4df1ede81cf48f7c35ca9ae3beeb Mon Sep 17 00:00:00 2001 From: amaliujia Date: Sat, 17 Oct 2026 20:48:03 +0800 Subject: [PATCH 2/8] Sync data and PutBlock before acknowledging a stream-committed PutBlock --- .../common/transport/server/ratis/DispatcherContext.java | 5 +++++ .../common/transport/server/ratis/LocalStream.java | 5 ++++- .../hadoop/ozone/container/keyvalue/KeyValueHandler.java | 6 ++++++ .../container/keyvalue/impl/KeyValueStreamDataChannel.java | 2 ++ .../transport/server/ratis/ContainerStateMachineTests.java | 1 + 5 files changed, 18 insertions(+), 1 deletion(-) diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/DispatcherContext.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/DispatcherContext.java index 3f72b0cb15f9..8e1140bc300a 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/DispatcherContext.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/DispatcherContext.java @@ -120,6 +120,11 @@ public boolean readFromTmpFile() { return this == READ_STATE_MACHINE_DATA; } + /** @return true if the PutBlock is committed through a data stream, i.e. without a Raft log entry. */ + public boolean isStreamCommit() { + return this == STREAM_LINK || this == STREAM_COMMAND; + } + public boolean validateToken() { switch (this) { case APPLY_TRANSACTION: 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 b2817e8bd3f0..f3a7ac41c973 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 @@ -64,7 +64,10 @@ public CompletableFuture onCommand(ByteBuffer buffer, long streamOff } return CompletableFuture.supplyAsync(() -> { try { - ((KeyValueStreamDataChannel) dataChannel).drainBuffers(); + final KeyValueStreamDataChannel channel = (KeyValueStreamDataChannel) dataChannel; + channel.drainBuffers(); + // Without a Raft log to replay, the data must be durable before the PutBlock is acknowledged. + channel.force(false); 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/KeyValueHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java index 8308c18804d6..a53343296233 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java @@ -699,6 +699,12 @@ ContainerCommandResponseProto handlePutBlock( dispatcherContext == null ? 0 : dispatcherContext.getLogIndex(); blockData.setBlockCommitSequenceId(bcsId); blockManager.putBlock(kvContainer, blockData, endOfBlock); + if (DispatcherContext.op(dispatcherContext).isStreamCommit()) { + // Without a Raft log to replay, the PutBlock must be durable before it is acknowledged. + try (DBHandle db = BlockUtils.getDB(kvContainer.getContainerData(), conf)) { + db.getStore().flushLog(true); + } + } blockDataProto = blockData.getProtoBufMessage(); 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 3ff82534f09d..392e9a7dc477 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 @@ -266,6 +266,8 @@ private static ContainerCommandRequestProto readPutBlockRequest(ByteBuffer b) private void closeWithStreamPutBlock() throws IOException { final ContainerCommandRequestProto proto = closeBuffers(buffers, super::writeFileChannel); + // Without a Raft log to replay, the data must be durable before the PutBlock is committed. + force(false); putBlockRequest.set(proto); Preconditions.checkState(putBlock != null); putBlock.accept(proto); 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 57cbae9b3c4b..b38891b48abc 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 @@ -267,6 +267,7 @@ public void testStreamCommandDrainsBuffersBeforePutBlock() throws Exception { final InOrder inOrder = inOrder(channel, dispatcher); inOrder.verify(channel).drainBuffers(); + inOrder.verify(channel).force(false); inOrder.verify(dispatcher).dispatch(argThat(r -> r.getCmdType() == ContainerProtos.Type.PutBlock), argThat(c -> DispatcherContext.op(c) == DispatcherContext.Op.STREAM_COMMAND)); } From 0a6366c58e045f92e31dc1364d8d2554e93c724f Mon Sep 17 00:00:00 2001 From: amaliujia Date: Mon, 26 Oct 2026 18:44:49 +0800 Subject: [PATCH 3/8] update --- .../hadoop/hdds/scm/OzoneClientConfig.java | 18 +++++++++ .../scm/storage/BlockDataStreamOutput.java | 37 ++++++++++++++++--- .../storage/TestBlockDataStreamOutput.java | 36 ++++++++++++++++++ .../server/ratis/ContainerStateMachine.java | 7 +++- .../server/ratis/DispatcherContext.java | 5 --- .../transport/server/ratis/LocalStream.java | 5 +-- .../container/keyvalue/KeyValueHandler.java | 3 +- .../impl/KeyValueStreamDataChannel.java | 2 - .../ratis/ContainerStateMachineTests.java | 16 +++++--- .../main/proto/DatanodeClientProtocol.proto | 2 + .../rpc/TestContainerStateMachineStream.java | 1 + 11 files changed, 106 insertions(+), 26 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java index 435478d1ac7c..b75547fa0bb3 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java @@ -348,6 +348,16 @@ public class OzoneClientConfig { tags = ConfigTag.CLIENT) private boolean datastreamPutBlockWithoutRaftEnabled = false; + @Config(key = "ozone.client.datastream.hsync.enabled", + defaultValue = "false", + type = ConfigType.BOOLEAN, + description = "When enabled, hsync of a Ratis data stream makes the datanodes sync the block data and " + + "the PutBlock to disk before replying. Otherwise, hsync returns once the datanodes have written them " + + "without sync. It takes effect only when hsync is enabled (ozone.fs.hsync.enabled) and the PutBlock " + + "is committed through the data stream (ozone.client.datastream.putblock.without.raft.enabled).", + tags = ConfigTag.CLIENT) + private boolean datastreamHsyncEnabled = false; + @Config(key = "ozone.client.key.write.concurrency", defaultValue = "1", description = "Maximum concurrent writes allowed on each key. " + @@ -779,6 +789,14 @@ public void setDatastreamPutBlockWithoutRaftEnabled(boolean datastreamPutBlockWi this.datastreamPutBlockWithoutRaftEnabled = datastreamPutBlockWithoutRaftEnabled; } + public boolean isDatastreamHsyncEnabled() { + return datastreamHsyncEnabled; + } + + public void setDatastreamHsyncEnabled(boolean datastreamHsyncEnabled) { + this.datastreamHsyncEnabled = datastreamHsyncEnabled; + } + /** * Enum for indicating what mode to use when combining chunk and block * checksums to define an aggregate FileChecksum. This should be considered diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java index 80b2417c93bf..85235fd3e3d4 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java @@ -147,6 +147,8 @@ public class BlockDataStreamOutput implements ByteBufferStreamOutput { private boolean isDatastreamPipelineMode; // Whether PutBlock is committed through the data stream instead of the Raft log; fixed for the stream lifetime. private final boolean putBlockWithoutRaft; + // Whether the most recent PutBlock command was not synced to disk by the datanodes. + private boolean unsyncedPutBlock = false; /** * Creates a new BlockDataStreamOutput. @@ -434,6 +436,14 @@ public void watchForCommit(boolean bufferFull) throws IOException { */ public void executePutBlock(boolean close, boolean force) throws IOException { + executePutBlock(close, force, false); + } + + /** + * @param sync whether the datanodes sync the data and the PutBlock to disk before replying; + * it applies only to a PutBlock committed in the middle of the stream without Raft + */ + private void executePutBlock(boolean close, boolean force, boolean sync) throws IOException { checkOpen(); long flushPos = totalDataFlushedLength; final List byteBufferList; @@ -477,7 +487,7 @@ public void executePutBlock(boolean close, return; } } else if (putBlockWithoutRaft) { - executePutBlockCommand(blockData, byteBufferList); + executePutBlockCommand(blockData, byteBufferList, sync); return; } try { @@ -534,13 +544,21 @@ public void executePutBlock(boolean close, * so far and the datanodes reply with the serialized {@link ContainerCommandResponseProto}. * * @param blockData the block metadata to commit - * @param byteBufferList the buffers covered by this PutBlock, to be released once it is acknowledged + * @param byteBufferList the buffers covered by this PutBlock, to be released once it is acknowledged; + * null if no data was written since the previous PutBlock + * @param sync whether the datanodes sync the data and the PutBlock to disk before replying */ private void executePutBlockCommand(BlockData blockData, - List byteBufferList) throws IOException { - final ContainerCommandRequestProto putBlockRequest + List byteBufferList, boolean sync) throws IOException { + ContainerCommandRequestProto putBlockRequest = ContainerProtocolCalls.getPutBlockRequest( xceiverClient.getPipeline(), blockData, false, tokenString); + if (sync) { + final ContainerCommandRequestProto.Builder builder = putBlockRequest.toBuilder(); + builder.getPutBlockBuilder().setSync(true); + putBlockRequest = builder.build(); + } + unsyncedPutBlock = !sync; final ByteBuffer command = ContainerCommandRequestMessage.toMessage( putBlockRequest, null).getContent().asReadOnlyByteBuffer(); Preconditions.checkState(command.remaining() <= PUT_BLOCK_REQUEST_LENGTH_MAX, @@ -559,7 +577,9 @@ private void executePutBlockCommand(BlockData blockData, } // The command has no log index; use 0 as for the standalone protocol // so that the buffers are released by the next watchForCommit. - commitWatcher.updateCommitInfoMap(0, byteBufferList); + if (byteBufferList != null) { + commitWatcher.updateCommitInfoMap(0, byteBufferList); + } }, responseExecutor)); } @@ -659,6 +679,8 @@ private void waitPutBlockCommandFuturesComplete() throws IOException { private void handleFlush(boolean close) throws IOException, InterruptedException, ExecutionException { checkOpen(); + // Only hsync flushes without closing. + final boolean sync = !close && config.isDatastreamHsyncEnabled(); // flush the last chunk data residing on the currentBuffer if (totalDataFlushedLength < writtenDataLength) { // This can be a partially filled chunk. Since we are flushing the buffer @@ -670,11 +692,14 @@ private void handleFlush(boolean close) currentBuffer = null; } updateFlushLength(); - executePutBlock(close, false); + executePutBlock(close, false, sync); } else if (close) { // forcing an "empty" putBlock if stream is being closed without new // data since latest flush - we need to send the "EOF" flag executePutBlock(true, true); + } else if (sync && putBlockWithoutRaft && unsyncedPutBlock) { + // hsync without new data: the data is committed by a previous PutBlock but not yet synced. + executePutBlock(false, true, true); } CompletableFuture.allOf(putBlockFutures.toArray(EMPTY_FUTURE_ARRAY)).get(); CompletableFuture.allOf(putBlockCommandFutures.toArray(EMPTY_FUTURE_ARRAY)).get(); diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java index c998e6e7ae73..3a215c054246 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java @@ -173,6 +173,42 @@ void midStreamPutBlockUsesStreamCommandWhenEnabled() throws Exception { assertArrayEquals(data, pipeline.getAllReceivedData()); } + @Test + void hsyncPutBlockCommandIsNotSyncedWhenDisabled() throws Exception { + MockDatanodePipeline pipeline = new MockDatanodePipeline(); + OzoneClientConfig config = createConfig(); + config.setDatastreamPutBlockWithoutRaftEnabled(true); + try (BlockDataStreamOutput stream = createStream(pipeline, config)) { + stream.write(ByteBuffer.wrap(randomBytes(400)), 0, 400); + stream.hsync(); + stream.write(ByteBuffer.wrap(randomBytes(50)), 0, 50); + stream.hsync(); + } + assertThat(pipeline.getReceivedCommands()).extracting(c -> c.getPutBlock().getSync()) + .containsExactly(false, false); + } + + @Test + void onlyHsyncPutBlockCommandIsSynced() throws Exception { + MockDatanodePipeline pipeline = new MockDatanodePipeline(); + OzoneClientConfig config = createConfig(); + config.setDatastreamPutBlockWithoutRaftEnabled(true); + config.setDatastreamHsyncEnabled(true); + try (BlockDataStreamOutput stream = createStream(pipeline, config)) { + // The flush boundary PutBlock is not synced. + stream.write(ByteBuffer.wrap(randomBytes(400)), 0, 400); + // hsync without new data sends a PutBlock only to sync the data committed by the boundary PutBlock. + stream.hsync(); + // hsync with new data syncs its own PutBlock. + stream.write(ByteBuffer.wrap(randomBytes(50)), 0, 50); + stream.hsync(); + // hsync after a synced PutBlock has nothing to sync. + stream.hsync(); + } + assertThat(pipeline.getReceivedCommands()).extracting(c -> c.getPutBlock().getSync()) + .containsExactly(false, true, true); + } + @Test void midStreamPutBlockCommandReleasesBuffers() throws Exception { MockDatanodePipeline pipeline = new MockDatanodePipeline(); diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java index 12def361e027..8dd67fb17dd6 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java @@ -752,7 +752,7 @@ void streamPutBlock(ContainerCommandRequestProto request) throws IOException { * @return the serialized {@link ContainerCommandResponseProto}, * which is identical on all the peers of the pipeline. */ - ByteBuffer streamCommand(ByteBuffer command) throws IOException { + ByteBuffer streamCommand(ByteBuffer command, DataChannel channel) throws IOException { // unsafeWrap is okay since the command buffer can be modified after this method has returned. final ContainerCommandRequestProto request = ContainerCommandRequestMessage.toProto( UnsafeByteOperations.unsafeWrap(command), getGroupId()); @@ -760,6 +760,9 @@ ByteBuffer streamCommand(ByteBuffer command) throws IOException { throw new StorageContainerException("Unexpected stream command " + request.getCmdType() + ", expected " + Type.PutBlock, ContainerProtos.Result.MALFORMED_REQUEST); } + if (request.getPutBlock().getSync()) { + channel.force(false); + } final DispatcherContext context = DispatcherContext.newBuilder(DispatcherContext.Op.STREAM_COMMAND) .setStage(DispatcherContext.WriteChunkStage.COMBINED) .setContainer2BCSIDMap(container2BCSIDMap) @@ -787,7 +790,7 @@ public CompletableFuture stream(RaftClientRequest request) { final DataChannel channel = getStreamDataChannel(requestProto, context); final ExecutorService chunkExecutor = requestProto.hasWriteChunk() ? getChunkExecutor(requestProto.getWriteChunk()) : null; - return new LocalStream(channel, chunkExecutor, this::streamCommand); + return new LocalStream(channel, chunkExecutor, command -> streamCommand(command, channel)); } catch (IOException e) { throw new CompletionException("Failed to create data stream", e); } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/DispatcherContext.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/DispatcherContext.java index 8e1140bc300a..3f72b0cb15f9 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/DispatcherContext.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/DispatcherContext.java @@ -120,11 +120,6 @@ public boolean readFromTmpFile() { return this == READ_STATE_MACHINE_DATA; } - /** @return true if the PutBlock is committed through a data stream, i.e. without a Raft log entry. */ - public boolean isStreamCommit() { - return this == STREAM_LINK || this == STREAM_COMMAND; - } - public boolean validateToken() { switch (this) { case APPLY_TRANSACTION: 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 f3a7ac41c973..b2817e8bd3f0 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 @@ -64,10 +64,7 @@ public CompletableFuture onCommand(ByteBuffer buffer, long streamOff } return CompletableFuture.supplyAsync(() -> { try { - final KeyValueStreamDataChannel channel = (KeyValueStreamDataChannel) dataChannel; - channel.drainBuffers(); - // Without a Raft log to replay, the data must be durable before the PutBlock is acknowledged. - channel.force(false); + ((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/KeyValueHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java index a53343296233..7f61065c782e 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java @@ -699,8 +699,7 @@ ContainerCommandResponseProto handlePutBlock( dispatcherContext == null ? 0 : dispatcherContext.getLogIndex(); blockData.setBlockCommitSequenceId(bcsId); blockManager.putBlock(kvContainer, blockData, endOfBlock); - if (DispatcherContext.op(dispatcherContext).isStreamCommit()) { - // Without a Raft log to replay, the PutBlock must be durable before it is acknowledged. + if (request.getPutBlock().getSync()) { try (DBHandle db = BlockUtils.getDB(kvContainer.getContainerData(), conf)) { db.getStore().flushLog(true); } 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 392e9a7dc477..3ff82534f09d 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 @@ -266,8 +266,6 @@ private static ContainerCommandRequestProto readPutBlockRequest(ByteBuffer b) private void closeWithStreamPutBlock() throws IOException { final ContainerCommandRequestProto proto = closeBuffers(buffers, super::writeFileChannel); - // Without a Raft log to replay, the data must be durable before the PutBlock is committed. - force(false); putBlockRequest.set(proto); Preconditions.checkState(putBlock != null); putBlock.accept(proto); 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 b38891b48abc..88a4228b7597 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 @@ -258,16 +258,17 @@ public void testWriteTimout() throws Exception { assertEquals(ContainerProtos.Result.CONTAINER_INTERNAL_ERROR, sce.getResult()); } - @Test - public void testStreamCommandDrainsBuffersBeforePutBlock() throws Exception { + @ParameterizedTest + @ValueSource(booleans = {false, true}) + public void testStreamCommandDrainsBuffersBeforePutBlock(boolean sync) throws Exception { final KeyValueStreamDataChannel channel = mock(KeyValueStreamDataChannel.class); final StateMachine.DataStream stream = setUpStream(channel); - stream.onCommand(newPutBlockCommand(), 0).get(); + stream.onCommand(newPutBlockCommand(sync), 0).get(); final InOrder inOrder = inOrder(channel, dispatcher); inOrder.verify(channel).drainBuffers(); - inOrder.verify(channel).force(false); + inOrder.verify(channel, times(sync ? 1 : 0)).force(false); inOrder.verify(dispatcher).dispatch(argThat(r -> r.getCmdType() == ContainerProtos.Type.PutBlock), argThat(c -> DispatcherContext.op(c) == DispatcherContext.Op.STREAM_COMMAND)); } @@ -311,11 +312,16 @@ private StateMachine.DataStream setUpStream(StateMachine.DataChannel channel) th } private static ByteBuffer newPutBlockCommand() { + return newPutBlockCommand(false); + } + + private static ByteBuffer newPutBlockCommand(boolean sync) { final ContainerProtos.ContainerCommandRequestProto putBlock = ContainerProtos.ContainerCommandRequestProto .newBuilder() .setCmdType(ContainerProtos.Type.PutBlock) .setPutBlock(ContainerProtos.PutBlockRequestProto.newBuilder() - .setBlockData(ContainerProtos.BlockData.newBuilder().setBlockID(newBlockID()).build())) + .setBlockData(ContainerProtos.BlockData.newBuilder().setBlockID(newBlockID()).build()) + .setSync(sync)) .setContainerID(1) .setDatanodeUuid(UUID.randomUUID().toString()) .build(); diff --git a/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto b/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto index 6b4979b3e2b7..fe8896474fd7 100644 --- a/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto +++ b/hadoop-hdds/interface-client/src/main/proto/DatanodeClientProtocol.proto @@ -342,6 +342,8 @@ message PutBlockRequestProto { required BlockData blockData = 1; optional bool eof = 2; optional bool containerAutoCreate = 3; + // Set by hsync; the datanode syncs the block file and the PutBlock to disk before replying. + optional bool sync = 4; } message PutBlockResponseProto { 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 5a7be96c44e5..588f1567c39a 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 @@ -158,6 +158,7 @@ void testPutBlockWithoutRaftCommittedOnAllDatanodes(boolean pipelineMode) throws OzoneConfiguration conf = new OzoneConfiguration(cluster().getConf()); OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); clientConfig.setDatastreamPutBlockWithoutRaftEnabled(true); + clientConfig.setDatastreamHsyncEnabled(true); clientConfig.setDatastreamPipelineMode(pipelineMode); conf.setFromObject(clientConfig); From dbbe9188156e4e6891a3bcf35fda7b2d596560c8 Mon Sep 17 00:00:00 2001 From: amaliujia Date: Mon, 26 Oct 2026 18:51:40 +0800 Subject: [PATCH 4/8] improve javadoc --- .../apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java | 3 +++ 1 file changed, 3 insertions(+) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java index 85235fd3e3d4..50b899083ccf 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java @@ -440,6 +440,9 @@ public void executePutBlock(boolean close, } /** + * @param close whether putBlock is happening as part of closing the stream + * @param force true if no data was written since most recent putBlock and + * either the stream is being closed or the previous putBlock is to be synced * @param sync whether the datanodes sync the data and the PutBlock to disk before replying; * it applies only to a PutBlock committed in the middle of the stream without Raft */ From 81f9ea60583169289e8f131c7d9cb50fead3593c Mon Sep 17 00:00:00 2001 From: amaliujia Date: Thu, 29 Oct 2026 11:47:01 +0800 Subject: [PATCH 5/8] update --- .../hadoop/hdds/scm/OzoneClientConfig.java | 18 ----- .../scm/storage/BlockDataStreamOutput.java | 72 +++++++------------ .../storage/TestBlockDataStreamOutput.java | 29 ++++---- .../scm/storage/ContainerProtocolCalls.java | 9 +-- .../storage/TestContainerProtocolCalls.java | 4 +- .../container/keyvalue/KeyValueContainer.java | 2 +- .../container/keyvalue/KeyValueHandler.java | 4 +- .../rpc/TestContainerStateMachineStream.java | 1 - 8 files changed, 48 insertions(+), 91 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java index b75547fa0bb3..435478d1ac7c 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/OzoneClientConfig.java @@ -348,16 +348,6 @@ public class OzoneClientConfig { tags = ConfigTag.CLIENT) private boolean datastreamPutBlockWithoutRaftEnabled = false; - @Config(key = "ozone.client.datastream.hsync.enabled", - defaultValue = "false", - type = ConfigType.BOOLEAN, - description = "When enabled, hsync of a Ratis data stream makes the datanodes sync the block data and " + - "the PutBlock to disk before replying. Otherwise, hsync returns once the datanodes have written them " + - "without sync. It takes effect only when hsync is enabled (ozone.fs.hsync.enabled) and the PutBlock " + - "is committed through the data stream (ozone.client.datastream.putblock.without.raft.enabled).", - tags = ConfigTag.CLIENT) - private boolean datastreamHsyncEnabled = false; - @Config(key = "ozone.client.key.write.concurrency", defaultValue = "1", description = "Maximum concurrent writes allowed on each key. " + @@ -789,14 +779,6 @@ public void setDatastreamPutBlockWithoutRaftEnabled(boolean datastreamPutBlockWi this.datastreamPutBlockWithoutRaftEnabled = datastreamPutBlockWithoutRaftEnabled; } - public boolean isDatastreamHsyncEnabled() { - return datastreamHsyncEnabled; - } - - public void setDatastreamHsyncEnabled(boolean datastreamHsyncEnabled) { - this.datastreamHsyncEnabled = datastreamHsyncEnabled; - } - /** * Enum for indicating what mode to use when combining chunk and block * checksums to define an aggregate FileChecksum. This should be considered diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java index 50b899083ccf..f8aa1f47edc2 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java @@ -147,8 +147,6 @@ public class BlockDataStreamOutput implements ByteBufferStreamOutput { private boolean isDatastreamPipelineMode; // Whether PutBlock is committed through the data stream instead of the Raft log; fixed for the stream lifetime. private final boolean putBlockWithoutRaft; - // Whether the most recent PutBlock command was not synced to disk by the datanodes. - private boolean unsyncedPutBlock = false; /** * Creates a new BlockDataStreamOutput. @@ -430,34 +428,17 @@ public void watchForCommit(boolean bufferFull) throws IOException { } /** + * @param sync whether the datanodes sync the data and the PutBlock to disk before replying; + * it applies only to a PutBlock committed in the middle of the stream without Raft * @param close whether putBlock is happening as part of closing the stream - * @param force true if no data was written since most recent putBlock and - * stream is being closed */ - public void executePutBlock(boolean close, - boolean force) throws IOException { - executePutBlock(close, force, false); - } - - /** - * @param close whether putBlock is happening as part of closing the stream - * @param force true if no data was written since most recent putBlock and - * either the stream is being closed or the previous putBlock is to be synced - * @param sync whether the datanodes sync the data and the PutBlock to disk before replying; - * it applies only to a PutBlock committed in the middle of the stream without Raft - */ - private void executePutBlock(boolean close, boolean force, boolean sync) throws IOException { + public void executePutBlock(boolean sync, boolean close) throws IOException { checkOpen(); long flushPos = totalDataFlushedLength; - final List byteBufferList; - if (!force) { - Objects.requireNonNull(bufferList, "bufferList == null"); - byteBufferList = buffersForPutBlock; - buffersForPutBlock = null; - Objects.requireNonNull(byteBufferList, "byteBufferList == null"); - } else { - byteBufferList = null; - } + Objects.requireNonNull(bufferList, "bufferList == null"); + // null if no data was written since the most recent putBlock + final List byteBufferList = buffersForPutBlock; + buffersForPutBlock = null; waitFuturesComplete(); if (close && putBlockWithoutRaft) { // Wait for boundary PutBlock(s) before appending the stream-close PutBlock. @@ -471,14 +452,14 @@ private void executePutBlock(boolean close, boolean force, boolean sync) throws // Then, old datanodes will fail since they expect a PutBlock. final ContainerCommandRequestProto putBlockRequest = ContainerProtocolCalls.getPutBlockRequest( - xceiverClient.getPipeline(), blockData, true, tokenString); + xceiverClient.getPipeline(), blockData, false, true, tokenString); dataStreamCloseReply = executePutBlockClose(putBlockRequest, PUT_BLOCK_REQUEST_LENGTH_MAX, out); dataStreamCloseReply.whenComplete((reply, e) -> { if (e != null || reply == null || !reply.isSuccess()) { LOG.warn("Failed executePutBlockClose, reply=" + reply, e); try { - executePutBlock(true, false); + executePutBlock(false, true); } catch (IOException ex) { throw new CompletionException(ex); } @@ -504,7 +485,7 @@ private void executePutBlock(boolean close, boolean force, boolean sync) throws throw new CompletionException(sce); } // if the ioException is not set, putBlock is successful - if (getIoException() == null && !force) { + if (getIoException() == null && byteBufferList != null) { BlockID responseBlockID = BlockID.getFromProtobuf( e.getPutBlock().getCommittedBlockLength().getBlockID()); Preconditions.checkState(blockID.get().getContainerBlockID() @@ -553,15 +534,9 @@ private void executePutBlock(boolean close, boolean force, boolean sync) throws */ private void executePutBlockCommand(BlockData blockData, List byteBufferList, boolean sync) throws IOException { - ContainerCommandRequestProto putBlockRequest + final ContainerCommandRequestProto putBlockRequest = ContainerProtocolCalls.getPutBlockRequest( - xceiverClient.getPipeline(), blockData, false, tokenString); - if (sync) { - final ContainerCommandRequestProto.Builder builder = putBlockRequest.toBuilder(); - builder.getPutBlockBuilder().setSync(true); - putBlockRequest = builder.build(); - } - unsyncedPutBlock = !sync; + xceiverClient.getPipeline(), blockData, sync, false, tokenString); final ByteBuffer command = ContainerCommandRequestMessage.toMessage( putBlockRequest, null).getContent().asReadOnlyByteBuffer(); Preconditions.checkState(command.remaining() <= PUT_BLOCK_REQUEST_LENGTH_MAX, @@ -636,14 +611,18 @@ public void flush() throws IOException { @Override public void hflush() throws IOException { - hsync(); + flushOrSync(false); } @Override public void hsync() throws IOException { + flushOrSync(true); + } + + private void flushOrSync(boolean sync) throws IOException { try { if (!isClosed()) { - handleFlush(false); + handleFlush(sync, false); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -677,13 +656,12 @@ private void waitPutBlockCommandFuturesComplete() throws IOException { } /** + * @param sync whether the datanodes sync the data and the PutBlock to disk before replying * @param close whether the flush is happening as part of closing the stream */ - private void handleFlush(boolean close) + private void handleFlush(boolean sync, boolean close) throws IOException, InterruptedException, ExecutionException { checkOpen(); - // Only hsync flushes without closing. - final boolean sync = !close && config.isDatastreamHsyncEnabled(); // flush the last chunk data residing on the currentBuffer if (totalDataFlushedLength < writtenDataLength) { // This can be a partially filled chunk. Since we are flushing the buffer @@ -695,14 +673,14 @@ private void handleFlush(boolean close) currentBuffer = null; } updateFlushLength(); - executePutBlock(close, false, sync); + executePutBlock(sync, close); } else if (close) { // forcing an "empty" putBlock if stream is being closed without new // data since latest flush - we need to send the "EOF" flag - executePutBlock(true, true); - } else if (sync && putBlockWithoutRaft && unsyncedPutBlock) { + executePutBlock(false, true); + } else if (sync && putBlockWithoutRaft) { // hsync without new data: the data is committed by a previous PutBlock but not yet synced. - executePutBlock(false, true, true); + executePutBlock(true, false); } CompletableFuture.allOf(putBlockFutures.toArray(EMPTY_FUTURE_ARRAY)).get(); CompletableFuture.allOf(putBlockCommandFutures.toArray(EMPTY_FUTURE_ARRAY)).get(); @@ -719,7 +697,7 @@ private void handleFlush(boolean close) public void close() throws IOException { if (xceiverClientFactory != null && xceiverClient != null) { try { - handleFlush(true); + handleFlush(false, true); dataStreamCloseReply.get(); } catch (ExecutionException e) { handleExecutionException(e); diff --git a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java index 3a215c054246..7d742d8d142e 100644 --- a/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java +++ b/hadoop-hdds/client/src/test/java/org/apache/hadoop/hdds/scm/storage/TestBlockDataStreamOutput.java @@ -174,15 +174,15 @@ void midStreamPutBlockUsesStreamCommandWhenEnabled() throws Exception { } @Test - void hsyncPutBlockCommandIsNotSyncedWhenDisabled() throws Exception { + void hflushPutBlockCommandIsNotSynced() throws Exception { MockDatanodePipeline pipeline = new MockDatanodePipeline(); OzoneClientConfig config = createConfig(); config.setDatastreamPutBlockWithoutRaftEnabled(true); try (BlockDataStreamOutput stream = createStream(pipeline, config)) { stream.write(ByteBuffer.wrap(randomBytes(400)), 0, 400); - stream.hsync(); + stream.hflush(); stream.write(ByteBuffer.wrap(randomBytes(50)), 0, 50); - stream.hsync(); + stream.hflush(); } assertThat(pipeline.getReceivedCommands()).extracting(c -> c.getPutBlock().getSync()) .containsExactly(false, false); @@ -193,20 +193,19 @@ void onlyHsyncPutBlockCommandIsSynced() throws Exception { MockDatanodePipeline pipeline = new MockDatanodePipeline(); OzoneClientConfig config = createConfig(); config.setDatastreamPutBlockWithoutRaftEnabled(true); - config.setDatastreamHsyncEnabled(true); - try (BlockDataStreamOutput stream = createStream(pipeline, config)) { - // The flush boundary PutBlock is not synced. - stream.write(ByteBuffer.wrap(randomBytes(400)), 0, 400); - // hsync without new data sends a PutBlock only to sync the data committed by the boundary PutBlock. - stream.hsync(); - // hsync with new data syncs its own PutBlock. - stream.write(ByteBuffer.wrap(randomBytes(50)), 0, 50); - stream.hsync(); - // hsync after a synced PutBlock has nothing to sync. - stream.hsync(); - } + BlockDataStreamOutput stream = createStream(pipeline, config); + // The flush boundary PutBlock is not synced. + stream.write(ByteBuffer.wrap(randomBytes(400)), 0, 400); + // hsync without new data sends a PutBlock only to sync the data committed by the boundary PutBlock. + stream.hsync(); + // hsync with new data syncs its own PutBlock. + stream.write(ByteBuffer.wrap(randomBytes(50)), 0, 50); + stream.hsync(); + stream.close(); + assertThat(pipeline.getReceivedCommands()).extracting(c -> c.getPutBlock().getSync()) .containsExactly(false, true, true); + assertEquals(450, stream.getTotalAckDataLength(), "The buffers of a synced PutBlock should be acknowledged"); } @Test diff --git a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java index b5be63d779f9..e76592d9c9ed 100644 --- a/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java +++ b/hadoop-hdds/common/src/main/java/org/apache/hadoop/hdds/scm/storage/ContainerProtocolCalls.java @@ -335,7 +335,7 @@ public static XceiverClientReply putBlockAsync(XceiverClientSpi xceiverClient, boolean containerAutoCreate) throws IOException, InterruptedException, ExecutionException { final ContainerCommandRequestProto request = getPutBlockRequest( - xceiverClient.getPipeline(), containerBlockData, eof, tokenString, + xceiverClient.getPipeline(), containerBlockData, false, eof, tokenString, containerAutoCreate); return xceiverClient.sendCommandAsync(request); } @@ -372,17 +372,18 @@ public static ContainerProtos.FinalizeBlockResponseProto finalizeBlock( } public static ContainerCommandRequestProto getPutBlockRequest( - Pipeline pipeline, BlockData containerBlockData, boolean eof, + Pipeline pipeline, BlockData containerBlockData, boolean sync, boolean eof, String tokenString) throws IOException { - return getPutBlockRequest(pipeline, containerBlockData, eof, tokenString, true); + return getPutBlockRequest(pipeline, containerBlockData, sync, eof, tokenString, true); } public static ContainerCommandRequestProto getPutBlockRequest( - Pipeline pipeline, BlockData containerBlockData, boolean eof, + Pipeline pipeline, BlockData containerBlockData, boolean sync, boolean eof, String tokenString, boolean containerAutoCreate) throws IOException { PutBlockRequestProto.Builder createBlockRequest = PutBlockRequestProto.newBuilder() .setBlockData(containerBlockData) + .setSync(sync) .setEof(eof); if (!containerAutoCreate) { createBlockRequest.setContainerAutoCreate(false); diff --git a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/storage/TestContainerProtocolCalls.java b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/storage/TestContainerProtocolCalls.java index 8d934c24e80f..e273029f9547 100644 --- a/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/storage/TestContainerProtocolCalls.java +++ b/hadoop-hdds/common/src/test/java/org/apache/hadoop/hdds/scm/storage/TestContainerProtocolCalls.java @@ -96,7 +96,7 @@ void putBlockRequestCarriesPipelineWriteVersion() throws IOException { .build(); ContainerCommandRequestProto request = - ContainerProtocolCalls.getPutBlockRequest(pipeline, blockData, true, null); + ContainerProtocolCalls.getPutBlockRequest(pipeline, blockData, false, true, null); assertEquals(VERSION, request.getWritePipelineVersion()); } @@ -164,7 +164,7 @@ void writeVersionPreservesOriginalSerializedValueFromWire() throws IOException { .build(); ContainerCommandRequestProto request = - ContainerProtocolCalls.getPutBlockRequest(pipeline, blockData, true, null); + ContainerProtocolCalls.getPutBlockRequest(pipeline, blockData, false, true, null); assertEquals(Integer.MAX_VALUE, request.getWritePipelineVersion()); } diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java index fd016f3078b9..6a7c2f674c61 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueContainer.java @@ -531,7 +531,7 @@ private void compactDB() throws StorageContainerException { } } - private void flushAndSyncDB() throws StorageContainerException { + void flushAndSyncDB() throws StorageContainerException { try { try (DBHandle db = BlockUtils.getDB(containerData, config)) { db.getStore().flushLog(true); diff --git a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java index 7f61065c782e..cdc459c96e22 100644 --- a/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java +++ b/hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/keyvalue/KeyValueHandler.java @@ -700,9 +700,7 @@ ContainerCommandResponseProto handlePutBlock( blockData.setBlockCommitSequenceId(bcsId); blockManager.putBlock(kvContainer, blockData, endOfBlock); if (request.getPutBlock().getSync()) { - try (DBHandle db = BlockUtils.getDB(kvContainer.getContainerData(), conf)) { - db.getStore().flushLog(true); - } + kvContainer.flushAndSyncDB(); } blockDataProto = blockData.getProtoBufMessage(); 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 588f1567c39a..5a7be96c44e5 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 @@ -158,7 +158,6 @@ void testPutBlockWithoutRaftCommittedOnAllDatanodes(boolean pipelineMode) throws OzoneConfiguration conf = new OzoneConfiguration(cluster().getConf()); OzoneClientConfig clientConfig = conf.getObject(OzoneClientConfig.class); clientConfig.setDatastreamPutBlockWithoutRaftEnabled(true); - clientConfig.setDatastreamHsyncEnabled(true); clientConfig.setDatastreamPipelineMode(pipelineMode); conf.setFromObject(clientConfig); From 6804504892ae096a6b2e4f7edf7930469f439345 Mon Sep 17 00:00:00 2001 From: amaliujia Date: Thu, 29 Oct 2026 12:45:48 +0800 Subject: [PATCH 6/8] fix patch --- .../hdds/scm/storage/BlockDataStreamOutput.java | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java index f8aa1f47edc2..798834a54d15 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java @@ -432,13 +432,18 @@ public void watchForCommit(boolean bufferFull) throws IOException { * it applies only to a PutBlock committed in the middle of the stream without Raft * @param close whether putBlock is happening as part of closing the stream */ - public void executePutBlock(boolean sync, boolean close) throws IOException { + private void executePutBlock(boolean sync, boolean close) throws IOException { checkOpen(); long flushPos = totalDataFlushedLength; - Objects.requireNonNull(bufferList, "bufferList == null"); - // null if no data was written since the most recent putBlock - final List byteBufferList = buffersForPutBlock; - buffersForPutBlock = null; + final List byteBufferList; + if (!sync && !close) { + Objects.requireNonNull(bufferList, "bufferList == null"); + byteBufferList = buffersForPutBlock; + buffersForPutBlock = null; + Objects.requireNonNull(byteBufferList, "byteBufferList == null"); + } else { + byteBufferList = null; + } waitFuturesComplete(); if (close && putBlockWithoutRaft) { // Wait for boundary PutBlock(s) before appending the stream-close PutBlock. @@ -485,7 +490,7 @@ public void executePutBlock(boolean sync, boolean close) throws IOException { throw new CompletionException(sce); } // if the ioException is not set, putBlock is successful - if (getIoException() == null && byteBufferList != null) { + if (getIoException() == null) { BlockID responseBlockID = BlockID.getFromProtobuf( e.getPutBlock().getCommittedBlockLength().getBlockID()); Preconditions.checkState(blockID.get().getContainerBlockID() From 0cccdf7cc951757de428e5686fe551928753bc43 Mon Sep 17 00:00:00 2001 From: amaliujia Date: Thu, 29 Oct 2026 13:06:24 +0800 Subject: [PATCH 7/8] expose ExecutePutBlock as public as there is other module caller --- .../apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java index 798834a54d15..032b80c866c0 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java @@ -432,7 +432,7 @@ public void watchForCommit(boolean bufferFull) throws IOException { * it applies only to a PutBlock committed in the middle of the stream without Raft * @param close whether putBlock is happening as part of closing the stream */ - private void executePutBlock(boolean sync, boolean close) throws IOException { + public void executePutBlock(boolean sync, boolean close) throws IOException { checkOpen(); long flushPos = totalDataFlushedLength; final List byteBufferList; From 62857a2b44b378d711cec229042b86e4249d06e3 Mon Sep 17 00:00:00 2001 From: amaliujia Date: Thu, 29 Oct 2026 14:10:45 +0800 Subject: [PATCH 8/8] continue iterating. --- .../hdds/scm/storage/BlockDataStreamOutput.java | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java index 032b80c866c0..9bf727841aa6 100644 --- a/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java +++ b/hadoop-hdds/client/src/main/java/org/apache/hadoop/hdds/scm/storage/BlockDataStreamOutput.java @@ -436,13 +436,14 @@ public void executePutBlock(boolean sync, boolean close) throws IOException { checkOpen(); long flushPos = totalDataFlushedLength; final List byteBufferList; - if (!sync && !close) { - Objects.requireNonNull(bufferList, "bufferList == null"); - byteBufferList = buffersForPutBlock; - buffersForPutBlock = null; - Objects.requireNonNull(byteBufferList, "byteBufferList == null"); + if (buffersForPutBlock != null) { + Objects.requireNonNull(bufferList, "bufferList == null"); + byteBufferList = buffersForPutBlock; + buffersForPutBlock = null; + Objects.requireNonNull(byteBufferList, "byteBufferList == null"); } else { - byteBufferList = null; + // no data was written since the most recent putBlock + byteBufferList = null; } waitFuturesComplete(); if (close && putBlockWithoutRaft) { @@ -490,7 +491,7 @@ public void executePutBlock(boolean sync, boolean close) throws IOException { throw new CompletionException(sce); } // if the ioException is not set, putBlock is successful - if (getIoException() == null) { + if (getIoException() == null && byteBufferList != null) { BlockID responseBlockID = BlockID.getFromProtobuf( e.getPutBlock().getCommittedBlockLength().getBlockID()); Preconditions.checkState(blockID.get().getContainerBlockID()