HDDS-16008. PutBlocks from Flushes also go without Raft

[amaliujia]

What changes were proposed in this pull request?

Apply https://issues.apache.org/jira/browse/RATIS-2625 to Ozone PutBlock commit. Use the new added Ratis data stream command API to commit PutBlock during flushes (which typically happens before the stream closes).

By doing so, once the config is enabled, all the PutBlock commits will be raftless.

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/HDDS-16008

How was this patch tested?

Unit test and integration test

[amaliujia]

The core changes are minors. Majority LOC is about testing.

[szetszwo]

@amaliujia , thanks for working on this! Please see the comments inlined.

[szetszwo]

We should add a new conf but not changing an existing conf. Otherwise, it becomes an incompatible change.

[amaliujia]

This feature is not released at all (for example in 2.1.2, there is no such code yet), so there is no compatibility that we need to maintain.

Then it is better that we only use one config for simplification.

[szetszwo]

Oh, you are right! Then, we should change it instead.

[szetszwo]

Let's just release the buffer here without using commitWatcher?

[amaliujia]

done. though I need to add a lock to byteBufferList in this case. You can tell from the diff how much mess I need to introduce now.

I feel like re-use commitWatcher.updateCommitInfoMap is still fine and have less complexity.

[szetszwo]

Then, let's use commitWatcher for now. We may simplify the code later on. Sorry that I was aware of the complication.

[amaliujia]

Sure revert back to use commitWatcher

[szetszwo]

Add a new ContainerCommandRequestMessage.toProto(ByteBuffer, RaftGroupId) method to avoid copying.

[amaliujia]

done

[szetszwo]

We may unsafeWrap(..) in streamCommand.

  ByteBuffer streamCommand(ByteBuffer command) 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());

[szetszwo]

@amaliujia , thanks for the update! Please see the responses above.

[szetszwo]

+1 the change looks good.

[peterxcli]

nit: wish we can refactor container SM and local stream, so we dont need to pass function as lambda. one reference-only refactor is -- LocalStream becomes the one place that handles a PutBlock arriving on the stream.

public class ContainerStateMachine extends BaseStateMachine {
    // ...
	CompletableFuture<DataStream> stream(RaftClientRequest request) {
	  init = toProto(request);
	  dispatchCommand(init, streamInitContext());          // StreamInit, checks the token for init's block
	  channel = (KeyValueStreamDataChannel) dispatcher.getStreamDataChannel(init);   // no callback parameter
	  return new LocalStream(channel, getChunkExecutor(init), init,
	      dispatcher, getGroupId(), container2BCSIDMap);
	}
	
	CompletableFuture<?> link(DataStream stream, LogEntryProto entry) {
	  return ((LocalStream) stream).link();               // V3: fails unless the close PutBlock was committed
	}

class LocalStream implements StateMachine.DataStream {
  KeyValueStreamDataChannel channel;
  DataChannel dataChannel;           // what Ratis writes to and closes
  BlockID blockID;                   // the block the stream was opened for
  ContainerDispatcher dispatcher; RaftGroupId groupId; Map<Long, Long> container2BCSIDMap; Executor executor;

  LocalStream(channel, executor, init, dispatcher, groupId, container2BCSIDMap) {
    blockID = init.getWriteChunk().getBlockID();
    dataChannel = init.getCmdType() == StreamInitWithPutBlock
        ? new PutBlockOnClose()      // V3: commit the trailing PutBlock at close
        : channel;                   // V2: the channel drops the trailing PutBlock (HDDS-12007)
  }

  DataChannel getDataChannel() { return dataChannel; }

  // PutBlock in the middle of the stream (flush / hsync)
  CompletableFuture<ByteBuffer> onCommand(ByteBuffer command, long offset) {
    return supplyAsync(() -> {
      channel.drainBuffers();
      return putBlock(toProto(command)).toByteString().asReadOnlyByteBuffer();
    }, executor);
  }

  // The single place that commits a PutBlock received on the stream
  private ContainerCommandResponseProto putBlock(ContainerCommandRequestProto request) {
    require request.getCmdType() == PutBlock;
    require blockIdOf(request) == blockID;
    context = DispatcherContext.newBuilder(Op.STREAM_PUT_BLOCK)
        .setStage(COMBINED).setContainer2BCSIDMap(container2BCSIDMap).build();
    response = dispatcher.dispatch(request, context);
    if (response.getResult() != SUCCESS) {
      throw new StorageContainerException(response.getMessage(), response.getResult());
    }
    return response;
  }

  // Ratis closes the stream through getDataChannel().close()
  private class PutBlockOnClose implements DataChannel {
    write(b)  -> channel.write(b);
    force(m)  -> channel.force(m);
    isOpen()  -> channel.isOpen();
    close() {
      trailing = channel.closeAndReadPutBlock();   // writes the remaining data
      putBlock(trailing);                          // throws on failure, so the stream close fails
      channel.setLinked();
    }
  }

  CompletableFuture<?> link() { /* checks from ContainerStateMachine.link() */ }
  cleanUp(), getExecutor()   // unchanged
}

maybe in another PR.

[amaliujia]

Sure we can keep this in mind and may refactor later.

[peterxcli]

Skipping this because sender has been verified first? so following message can be trusted unconditionaly?

[amaliujia]

This is used for dispatch context but not the stream header:

    final DispatcherContext context = DispatcherContext.newBuilder(DispatcherContext.Op.STREAM_COMMAND)
        .setStage(DispatcherContext.WriteChunkStage.COMBINED)
        .setContainer2BCSIDMap(container2BCSIDMap)
        .build();

[peterxcli]

Thanks! LGTM +1.