RATIS-2625. Streaming: support sending commands in Data Stream by amaliujia · Pull Request #1534 · apache/ratis

amaliujia · 2026-07-29T02:48:51Z

What changes were proposed in this pull request?

This PR is AI-assisted

Adds a first-class mid-stream command plane to Ratis data streams. Callers can send opaque command payloads during an active stream via DataStreamOutput.commandAsync(). The server delivers them to StateMachine.DataStream.onCommand() instead of writing bytes to the data channel. Commands are ordered with data writes, replicated to remote peers like STREAM_DATA, and do not advance the stream byte offset.

Example usage:

// Client
out.getHeaderFuture().join();
out.writeAsync(dataBuffer).join();
out.commandAsync(myCommandBytes).join();  // opaque payload
out.writeAsync(moreData, StandardWriteOption.CLOSE).join();

// State machine
@Override
public CompletableFuture<?> onCommand(ByteBuffer command, long streamOffset) {
  // parse command, e.g. force(), metadata update, etc.
  return CompletableFuture.completedFuture(null);
}

What is the link to the Apache JIRA

https://issues.apache.org/jira/browse/RATIS-2625

How was this patch tested?

Unit tests

amaliujia · 2026-07-30T06:23:06Z

@szetszwo this is the follow up for apache/ozone#10823 (comment)

szetszwo

@amaliujia , thanks for working on this! The change looks quite good.

The StateMachine may return a result, we should send it back to the client; see https://issues.apache.org/jira/secure/attachment/13083627/1534_review2.patch

szetszwo · 2026-07-30T21:31:32Z

+     * @param streamOffset the current stream byte offset
+     * @return a future for the command task
+     */
+    default CompletableFuture<?> onCommand(ByteBuffer command, long streamOffset) {


Since this is a commnad, it should return CompletableFuture<ByteBuffer>.

szetszwo

@amaliujia , thanks for the update! Please see the comments inlined.

szetszwo · 2026-07-31T17:20:31Z


+    @Override
+    public CompletableFuture<DataStreamReply> commandAsync(ByteBuffer src) {
+      return commandAsyncImpl(src, src.remaining());


commandAsyncImpl is used just once. Let's inline the code.

public CompletableFuture<DataStreamReply> commandAsync(ByteBuffer src) { if (isClosed()) { return JavaUtils.completeExceptionally(new AlreadyClosedException( clientId + ": stream already closed, request=" + header)); } return combineHeader(send(Type.STREAM_COMMAND, src, src.remaining(), Collections.singleton(StandardWriteOption.FLUSH))); }

szetszwo · 2026-07-31T17:22:27Z

    private final CompletableFuture<DataStream> streamFuture;
-    private final AtomicReference<CompletableFuture<Long>> writeFuture;
-    private final RequestMetrics metrics;
+    private final AtomicReference<CompletableFuture<LocalResult>> writeFuture;


Let's rename it to resultFuture.

szetszwo · 2026-07-31T17:23:10Z

+    CompletableFuture<LocalResult> write(ByteBuf buf, Iterable<WriteOption> options,
                                  Executor executor) {


Make it a single line.

szetszwo · 2026-07-31T17:26:07Z

+      final ByteBuffer command = copyBuffer(request.slice());
+      localResult = info.getLocal().command(command, request.getStreamOffset(), writeExecutor);
+      remoteWrites = info.applyToRemotes(out -> out.command(
+          copyBuffer(request.slice()), requestExecutor));


duplicate() and reuse command, i.e.

final ByteBuffer command = copyBuffer(request.slice()); localResult = info.getLocal().command(command.duplicate(), request.getStreamOffset(), writeExecutor); remoteWrites = info.applyToRemotes(out -> out.command(command, requestExecutor));

szetszwo · 2026-07-31T17:27:02Z

+      copy.put(buffer);
+    }
+    copy.flip();
+    return copy;


Make it readonly

return copy.asReadOnlyBuffer();

szetszwo · 2026-07-31T17:31:48Z

+    return CompletableFuture.runAsync(() -> {}, e)
+        .thenCompose(ignored -> stream.onCommand(command, streamOffset));


Using completedFuture(null) is more efficient since it doesn't have to run an empty task.

return CompletableFuture.completedFuture(null) .thenComposeAsync(ignored -> stream.onCommand(command, streamOffset), e);

Thanks. I really learned something from this comment.

szetszwo · 2026-07-31T17:34:56Z

        reply = replyFuture.get(requestTimeout.getDuration(), requestTimeout.getUnit());
      } catch (Exception e) {
-        throw new CompletionException("Failed to get reply for bytesWritten=" + bytesWritten + ", " + request, e);
+        throw new CompletionException("Failed to get reply for " + localResult + ", " + request, e);


Add toString()

@Override public String toString() { return commandReply != null ? "commandReply:" + StringUtils.bytes2HexString(commandReply) : "byteWritten:" + byteWritten; }

ss77892 · 2026-08-03T05:39:19Z

    }

+    @Override
+    public CompletableFuture<DataStreamReply> commandAsync(ByteBuffer src) {


What would happen on the client side if it issues two commands without data in between? As far as I can see, NettyClientReply, when it maps requests, considers only the stream offset and type, so the second command with the same offset wouldn't have a ReplyEntry.

Very good call. It seems to me that the first command's ReplyEntry will be used for the second command and the second command's RequestEntry won't be in the map. Anyway it looks like a mess without a handling.

So I instead fail the commands if there is already one at the same stream offset. After all, we do not expect the caller issue multiple commands at the same stream offset for now.

szetszwo

@ss77892 , Good catch one the problem!

@amaliujia , thanks for the update! Please see the comment inlined.

szetszwo · 2026-08-03T22:02:53Z

+      if (requestEntry.type == Type.STREAM_COMMAND) {
+        if (map.containsKey(requestEntry)) {
+          final IllegalStateException exception = new IllegalStateException(
+              "A STREAM_COMMAND is already pending for " + requestEntry
+              + " for " + clientInvocationId
+              + "; wait for the previous command reply before sending another");
+          f.completeExceptionally(exception);
+          return null;
+        }
+        final ReplyEntry entry = new ReplyEntry(isClose, f);
+        map.put(requestEntry, entry);
+        return entry;
+      }
      // ConcurrentHashMap.computeIfAbsent javadoc: the function is applied at most once per key.
      return map.computeIfAbsent(requestEntry, r -> new ReplyEntry(isClose, f));


ReplyEntry submitRequest(RequestEntry requestEntry, boolean isClose, CompletableFuture<DataStreamReply> f) { LOG.debug("put {} to the map for {}", requestEntry, clientInvocationId); final MemoizedSupplier<ReplyEntry> supplier = MemoizedSupplier.valueOf(() -> new ReplyEntry(isClose, f)); final ReplyEntry reply = map.computeIfAbsent(requestEntry, r -> supplier.get()); if (requestEntry.type == Type.STREAM_COMMAND && !supplier.isInitialized()) { final IllegalStateException exception = new IllegalStateException( "STREAM_COMMAND already exist: " + requestEntry + " for " + clientInvocationId); f.completeExceptionally(exception); return null; } return reply; }

We should only make one call to the map as above; otherwise, it is not atomic. In this case, we may use MemoizedSupplier.

Let's also make the exception message shorter; see below

Done!

Thanks for the hint. The map is concurrent map so two accesses (one containsKey, one put) are not atomic and will cause concurrency issue.

szetszwo · 2026-08-03T22:23:32Z

RATIS-2625. Streaming: support control plane in Data Stream

Control plane seems not the right term. How about revising the title to below?

Streaming: support sending commands in Data Stream

szetszwo

+1 the change looks good.

update

ab19857

amaliujia marked this pull request as draft July 29, 2026 02:48

amaliujia added 4 commits July 29, 2026 10:55

fix

2cac57a

update

14fc6c3

update

eea3b1b

update

00fd13b

amaliujia changed the title ~~update~~ [In Progress] Control Plane in Data Stream API Jul 29, 2026

amaliujia added 2 commits July 30, 2026 10:31

update

c2a7221

remove filestore change

ffad563

amaliujia changed the title ~~[In Progress] Control Plane in Data Stream API~~ RATIS-2625. Control Plane in Data Stream API Jul 30, 2026

amaliujia changed the title ~~RATIS-2625. Control Plane in Data Stream API~~ RATIS-2625. Streaming: support control plane in Data Stream Jul 30, 2026

amaliujia marked this pull request as ready for review July 30, 2026 03:28

update

29d0df7

szetszwo reviewed Jul 30, 2026

View reviewed changes

amaliujia added 2 commits July 31, 2026 12:16

address review comments

51d0d14

update

3fcff8d

szetszwo reviewed Jul 31, 2026

View reviewed changes

amaliujia added 2 commits August 3, 2026 11:42

address comments

2b80b08

fix findsBug

1b69a7a

ss77892 reviewed Aug 3, 2026

View reviewed changes

address comments

d417115

szetszwo reviewed Aug 3, 2026

View reviewed changes

amaliujia changed the title ~~RATIS-2625. Streaming: support control plane in Data Stream~~ RATIS-2625. Streaming: support sending commands in Data Stream Aug 4, 2026

Make one call to the map

02ee091

szetszwo approved these changes Aug 4, 2026

View reviewed changes

szetszwo merged commit 64e5b34 into apache:master Aug 4, 2026
16 checks passed

		CompletableFuture<LocalResult> write(ByteBuf buf, Iterable<WriteOption> options,
		Executor executor) {

		return CompletableFuture.runAsync(() -> {}, e)
		.thenCompose(ignored -> stream.onCommand(command, streamOffset));

Uh oh!

Conversation

amaliujia commented Jul 29, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What changes were proposed in this pull request?

What is the link to the Apache JIRA

How was this patch tested?

Uh oh!

amaliujia commented Jul 30, 2026

Uh oh!

szetszwo left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szetszwo Jul 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szetszwo left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

amaliujia Aug 3, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szetszwo left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

szetszwo commented Aug 3, 2026

Uh oh!

szetszwo left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

amaliujia commented Jul 29, 2026 •

edited

Loading

szetszwo left a comment •

edited

Loading

szetszwo Jul 30, 2026 •

edited

Loading

amaliujia Aug 3, 2026 •

edited

Loading