peterxcli commented on code in PR #11356:
URL: https://github.com/apache/ozone/pull/11356#discussion_r4192673719
##########
hadoop-hdds/container-service/src/main/java/org/apache/hadoop/ozone/container/common/transport/server/ratis/ContainerStateMachine.java:
##########
@@ -740,6 +742,33 @@ void streamPutBlock(ContainerCommandRequestProto request)
throws IOException {
dispatchCommand(request, context);
}
+ /**
+ * Commit a PutBlock sent as a command in the middle of a data stream;
+ * see {@link StateMachine.DataStream#onCommand(ByteBuffer, long)}.
+ * The PutBlock is applied without a Raft log entry, so it has no log index
to use as bcsId.
+ *
+ * @return the serialized {@link ContainerCommandResponseProto},
+ * which is identical on all the peers of the pipeline.
+ */
+ ByteBuffer streamCommand(ByteBuffer command) throws IOException {
Review Comment:
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.
```java
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
}
```
```java
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
