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..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 @@ -428,21 +428,21 @@ 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 { + public void executePutBlock(boolean sync, boolean close) throws IOException { checkOpen(); long flushPos = totalDataFlushedLength; final List byteBufferList; - if (!force) { + if (buffersForPutBlock != null) { Objects.requireNonNull(bufferList, "bufferList == null"); byteBufferList = buffersForPutBlock; buffersForPutBlock = null; Objects.requireNonNull(byteBufferList, "byteBufferList == null"); } else { + // no data was written since the most recent putBlock byteBufferList = null; } waitFuturesComplete(); @@ -458,14 +458,14 @@ public void executePutBlock(boolean close, // 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); } @@ -477,7 +477,7 @@ public void executePutBlock(boolean close, return; } } else if (putBlockWithoutRaft) { - executePutBlockCommand(blockData, byteBufferList); + executePutBlockCommand(blockData, byteBufferList, sync); return; } try { @@ -491,7 +491,7 @@ public void executePutBlock(boolean close, 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() @@ -534,13 +534,15 @@ 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 { + List byteBufferList, boolean sync) throws IOException { final ContainerCommandRequestProto putBlockRequest = ContainerProtocolCalls.getPutBlockRequest( - xceiverClient.getPipeline(), blockData, false, tokenString); + xceiverClient.getPipeline(), blockData, sync, false, tokenString); final ByteBuffer command = ContainerCommandRequestMessage.toMessage( putBlockRequest, null).getContent().asReadOnlyByteBuffer(); Preconditions.checkState(command.remaining() <= PUT_BLOCK_REQUEST_LENGTH_MAX, @@ -559,7 +561,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)); } @@ -613,14 +617,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(); @@ -654,9 +662,10 @@ 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(); // flush the last chunk data residing on the currentBuffer @@ -670,11 +679,14 @@ private void handleFlush(boolean close) currentBuffer = null; } updateFlushLength(); - executePutBlock(close, false); + 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); + executePutBlock(false, true); + } else if (sync && putBlockWithoutRaft) { + // hsync without new data: the data is committed by a previous PutBlock but not yet synced. + executePutBlock(true, false); } CompletableFuture.allOf(putBlockFutures.toArray(EMPTY_FUTURE_ARRAY)).get(); CompletableFuture.allOf(putBlockCommandFutures.toArray(EMPTY_FUTURE_ARRAY)).get(); @@ -691,7 +703,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 c998e6e7ae73..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 @@ -173,6 +173,41 @@ void midStreamPutBlockUsesStreamCommandWhenEnabled() throws Exception { assertArrayEquals(data, pipeline.getAllReceivedData()); } + @Test + 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.hflush(); + stream.write(ByteBuffer.wrap(randomBytes(50)), 0, 50); + stream.hflush(); + } + 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); + 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 void midStreamPutBlockCommandReleasesBuffers() throws Exception { MockDatanodePipeline pipeline = new MockDatanodePipeline(); 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/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/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 8308c18804d6..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 @@ -699,6 +699,9 @@ ContainerCommandResponseProto handlePutBlock( dispatcherContext == null ? 0 : dispatcherContext.getLogIndex(); blockData.setBlockCommitSequenceId(bcsId); blockManager.putBlock(kvContainer, blockData, endOfBlock); + if (request.getPutBlock().getSync()) { + kvContainer.flushAndSyncDB(); + } blockDataProto = blockData.getProtoBufMessage(); 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..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,15 +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, 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)); } @@ -310,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 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()); + } + } }