HDDS-17093. Sync block data and PutBlock before acknowledging a stream-committed PutBlock

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<StreamBuffer> 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<StreamBuffer> byteBufferList) throws IOException {
+      List<StreamBuffer> 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);

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();

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);

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());
   }

hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java

@@ -752,14 +752,17 @@ 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());
     if (request.getCmdType() != Type.PutBlock) {
       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<DataStream> 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);
       }

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);

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();
 

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();

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 {