RATIS-2625. Streaming: support control plane 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);

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

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

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

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

RATIS-2625. Streaming: support control plane in Data Stream#1534
amaliujia wants to merge 10 commits into
apache:masterfrom
amaliujia:control_api

amaliujia commented Jul 29, 2026 •

edited

Loading

Uh oh!

amaliujia commented Jul 30, 2026

Uh oh!

szetszwo left a comment •

edited

Loading

Uh oh!

szetszwo Jul 30, 2026 •

edited

Loading

Uh oh!

szetszwo left a comment

Uh oh!

szetszwo Jul 31, 2026

Uh oh!

szetszwo Jul 31, 2026

Uh oh!

szetszwo Jul 31, 2026

Uh oh!

szetszwo Jul 31, 2026

Uh oh!

szetszwo Jul 31, 2026

Uh oh!

szetszwo Jul 31, 2026

Uh oh!

szetszwo Jul 31, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

		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!

szetszwo Jul 31, 2026

Choose a reason for hiding this comment

Uh oh!

szetszwo Jul 31, 2026

Choose a reason for hiding this comment

Uh oh!

szetszwo Jul 31, 2026

Choose a reason for hiding this comment

Uh oh!

szetszwo Jul 31, 2026

Choose a reason for hiding this comment

Uh oh!

szetszwo Jul 31, 2026

Choose a reason for hiding this comment

Uh oh!

szetszwo Jul 31, 2026

Choose a reason for hiding this comment

Uh oh!

szetszwo Jul 31, 2026

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

amaliujia commented Jul 29, 2026 •

edited

Loading

szetszwo left a comment •

edited

Loading

szetszwo Jul 30, 2026 •

edited

Loading