Refactor tri pack & unpack use stream by EarthChen · Pull Request #16045 · apache/dubbo

EarthChen · 2026-01-23T11:02:38Z

What is the purpose of the change?

The current PR is based on #16041, and PR #16041 needs to be merged first.

Checklist

Make sure there is a GitHub_issue field for the change.
Write a pull request description that is detailed enough to understand what the pull request does, how, and why.
Write necessary unit-test to verify your logic correction. If the new feature or significant change is committed, please remember to add sample in dubbo samples project.
Make sure gitHub actions can pass. Why the workflow is failing and how to fix it?

…streams

…dubbo/remoting/http12/h2/Http2ServerChannelObserver.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

…eam and Http2ServerChannelObserver

…bservers

…reamingDecoder

This commit unifies the deframer implementation between client and server sides for the Triple protocol, eliminating code duplication and improving maintainability. ## Key Changes ### Deleted Files - TriDecoder.java: Removed client-specific deframer implementation - Deframer.java: Removed client deframer interface - RecordListener.java: Removed test helper - TriDecoderTest.java: Removed corresponding test ### Core Modifications 1. **LengthFieldStreamingDecoder**: Added BoundedInputStream to support: - Message boundary isolation (prevents reading beyond current message) - mark/reset support required by Hessian2 and other deserializers - Zero-copy optimization by reading directly from accumulate stream 2. **StreamingDecoder.FragmentListener**: Added messageLength parameter to onFragmentMessage() method for better flow control 3. **AbstractTripleClientStream**: Migrated from TriDecoder to GrpcStreamingDecoder, using ByteBufInputStream to adapt Netty's ByteBuf to InputStream 4. **Stream.Listener.onMessage**: Changed parameter from byte[] to InputStream for unified handling and memory optimization 5. **PackableMethod**: Added default parseResponse(InputStream) method for backward compatibility 6. **CompositeInputStream**: Improved stream release logic to prevent exceptions 7. **DescriptorUtils**: Added mark() call before reading stream to support reset ## Architecture Change Before: - Client: ByteBuf -> TriDecoder (Netty-specific) -> byte[] -> deserializer - Server: InputStream -> LengthFieldStreamingDecoder -> byte[] -> deserializer After: - Client: ByteBuf -> ByteBufInputStream -> GrpcStreamingDecoder -> BoundedInputStream -> deserializer - Server: InputStream -> GrpcStreamingDecoder -> BoundedInputStream -> deserializer ## Bug Fix Fixed BoundedInputStream.reset() not restoring the 'remaining' counter, which caused streams to return EOF after mark/reset. This was the root cause of POJO method invocation failures with "Unexpected serialization type:null" error.

…putStream

codecov-commenter · 2026-01-23T11:25:53Z

Codecov Report

❌ Patch coverage is 60.13986% with 57 lines in your changes missing coverage. Please review.
✅ Project coverage is 60.74%. Comparing base (f27d0d7) to head (bb6d634).

Files with missing lines	Patch %	Lines
...pc/protocol/tri/TripleCustomerProtocolWrapper.java	54.02%	31 Missing and 9 partials ⚠️
...java/org/apache/dubbo/rpc/model/WrapperUnPack.java	16.66%	5 Missing ⚠️
...g/apache/dubbo/rpc/protocol/tri/PbArrayPacker.java	0.00%	4 Missing ⚠️
...bbo/rpc/protocol/tri/ReflectionPackableMethod.java	81.81%	4 Missing ⚠️
...ava/org/apache/dubbo/rpc/model/PackableMethod.java	70.00%	3 Missing ⚠️
...apache/dubbo/rpc/protocol/tri/DescriptorUtils.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##                3.3   #16045      +/-   ##
============================================
- Coverage     60.75%   60.74%   -0.02%     
- Complexity    11751    11755       +4     
============================================
  Files          1951     1953       +2     
  Lines         88992    89120     +128     
  Branches      13418    13443      +25     
============================================
+ Hits          54070    54133      +63     
- Misses        29362    29423      +61     
- Partials       5560     5564       +4

Flag	Coverage Δ
integration-tests-java21	`32.12% <57.34%> (-0.07%)`	⬇️
integration-tests-java8	`32.24% <57.34%> (-0.07%)`	⬇️
samples-tests-java21	`32.15% <31.46%> (-0.04%)`	⬇️
samples-tests-java8	`29.84% <30.76%> (-0.03%)`	⬇️
unit-tests-java11	`58.98% <58.74%> (-0.03%)`	⬇️
unit-tests-java17	`58.48% <58.74%> (-0.02%)`	⬇️
unit-tests-java21	`58.48% <58.74%> (-0.03%)`	⬇️
unit-tests-java25	`58.43% <58.74%> (-0.05%)`	⬇️
unit-tests-java8	`58.99% <58.74%> (-0.04%)`	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

Copilot

Pull request overview

This PR refactors the Triple protocol's pack and unpack operations to use streams (InputStream/OutputStream) instead of byte arrays. This change improves memory efficiency by avoiding unnecessary buffering of large messages into memory, and enables more efficient streaming processing.

Changes:

Replaced byte array-based serialization/deserialization APIs with stream-based APIs in Pack/UnPack/PackableMethod interfaces
Refactored TriDecoder to use GrpcStreamingDecoder (unified with server-side implementation)
Updated all Pack/UnPack implementations to support the new stream-based APIs with backward compatibility
Introduced BoundedInputStream for safe stream reading with mark/reset support

Reviewed changes

Copilot reviewed 26 out of 26 changed files in this pull request and generated 11 comments.

Show a summary per file

File	Description
Stream.java	Updated Listener.onMessage to accept InputStream with messageLength parameter
AbstractTripleClientStream.java	Replaced TriDecoder with GrpcStreamingDecoder and updated to use ByteBufInputStream
GrpcStreamingDecoder.java	Implements stream-based message decompression
GrpcCompositeCodec.java	Updated encode/decode to use streams instead of byte arrays
ReflectionPackableMethod.java	Implemented stream-based pack/unpack methods with backward compatibility
TripleCustomerProtocolWrapper.java	Added parseFrom(InputStream) and writeTo(OutputStream) methods with proper varint and field reading
PbUnpack.java & PbArrayPacker.java	Implemented stream-based unpack/pack methods
LengthFieldStreamingDecoder.java	Added BoundedInputStream for controlled stream reading with mark/reset support
Pack.java & UnPack.java	Added new stream-based methods with default implementations and deprecated byte array methods
PackableMethod.java	Added stream-based parseRequest/Response and packRequest/Response methods
MockClientStreamListener.java	Updated test mock to handle InputStream parameters
TriDecoderTest.java & RecordListener.java	Removed obsolete tests for deleted TriDecoder

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

Copilot · 2026-01-24T11:38:01Z

...pc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleCustomerProtocolWrapper.java

+                int tag = b;
+                if ((b & 0x80) != 0) {
+                    int shift = 7;
+                    tag = b & 0x7F;
+                    while (shift < 35) {
+                        b = inputStream.read();
+                        if (b == -1) {
+                            throw new IOException("Unexpected end of stream while reading tag");
+                        }
+                        tag |= (b & 0x7F) << shift;
+                        if ((b & 0x80) == 0) {
+                            break;
+                        }
+                        shift += 7;
+                    }
+                }
+


The varint parsing logic for tags (lines 208-223) is duplicated and slightly different from the readRawVarint32 method. Consider using readRawVarint32 for both tag and length parsing to reduce duplication and ensure consistency. The current implementation reads the first byte outside the loop and then continues if needed, which could be simplified.

Suggested change

int tag = b;

if ((b & 0x80) != 0) {

int shift = 7;

tag = b & 0x7F;

while (shift < 35) {

b = inputStream.read();

if (b == -1) {

throw new IOException("Unexpected end of stream while reading tag");

}

tag |= (b & 0x7F) << shift;

if ((b & 0x80) == 0) {

break;

}

shift += 7;

}

}

final int firstByte = b;

InputStream tagInputStream = new InputStream() {

private boolean first = true;

@Override

public int read() throws IOException {

if (first) {

first = false;

return firstByte;

}

return inputStream.read();

}

@Override

public int read(byte[] buffer, int off, int len) throws IOException {

if (buffer == null) {

throw new NullPointerException();

}

if (off < 0 || len < 0 || len > buffer.length - off) {

throw new IndexOutOfBoundsException();

}

if (len == 0) {

return 0;

}

if (first) {

first = false;

buffer[off] = (byte) firstByte;

return 1;

}

return inputStream.read(buffer, off, len);

}

};

int tag = readRawVarint32(tagInputStream);

Copilot · 2026-01-24T11:38:02Z

...pc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleCustomerProtocolWrapper.java

+                    tripleResponseWrapper.serializeType = new String(fieldData);
+                } else if (fieldNum == 2) {
+                    tripleResponseWrapper.data = fieldData;
+                } else if (fieldNum == 3) {
+                    tripleResponseWrapper.type = new String(fieldData);


String construction should specify charset explicitly. The current code uses the default platform charset (via new String(byte[])) which may differ across systems. Consider using new String(fieldData, StandardCharsets.UTF_8) to match the encoding used in toByteArray().

Suggested change

tripleResponseWrapper.serializeType = new String(fieldData);

} else if (fieldNum == 2) {

tripleResponseWrapper.data = fieldData;

} else if (fieldNum == 3) {

tripleResponseWrapper.type = new String(fieldData);

tripleResponseWrapper.serializeType = new String(fieldData, StandardCharsets.UTF_8);

} else if (fieldNum == 2) {

tripleResponseWrapper.data = fieldData;

} else if (fieldNum == 3) {

tripleResponseWrapper.type = new String(fieldData, StandardCharsets.UTF_8);

Copilot · 2026-01-24T11:38:02Z

...pc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleCustomerProtocolWrapper.java

+                    tripleRequestWrapper.serializeType = new String(fieldData);
+                } else if (fieldNum == 2) {
+                    tripleRequestWrapper.args.add(fieldData);
+                } else if (fieldNum == 3) {
+                    tripleRequestWrapper.argTypes.add(new String(fieldData));


String construction should specify charset explicitly. The current code uses the default platform charset (via new String(byte[])) which may differ across systems. Consider using new String(fieldData, StandardCharsets.UTF_8) to match the encoding used in toByteArray().

Suggested change

tripleRequestWrapper.serializeType = new String(fieldData);

} else if (fieldNum == 2) {

tripleRequestWrapper.args.add(fieldData);

} else if (fieldNum == 3) {

tripleRequestWrapper.argTypes.add(new String(fieldData));

tripleRequestWrapper.serializeType = new String(fieldData, StandardCharsets.UTF_8);

} else if (fieldNum == 2) {

tripleRequestWrapper.args.add(fieldData);

} else if (fieldNum == 3) {

tripleRequestWrapper.argTypes.add(new String(fieldData, StandardCharsets.UTF_8));

...-triple/src/test/java/org/apache/dubbo/rpc/protocol/tri/stream/MockClientStreamListener.java

...tp12/src/main/java/org/apache/dubbo/remoting/http12/message/LengthFieldStreamingDecoder.java

Copilot · 2026-01-24T11:38:03Z

dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/DescriptorUtils.java

@@ -125,9 +124,9 @@ public static MethodDescriptor findTripleMethodDescriptor(
            ServiceDescriptor serviceDescriptor, String methodName, InputStream rawMessage) throws IOException {
        MethodDescriptor methodDescriptor = findReflectionMethodDescriptor(serviceDescriptor, methodName);
        if (methodDescriptor == null) {


Using mark with Integer.MAX_VALUE as the readlimit parameter may not work with all InputStream implementations. The BoundedInputStream created in LengthFieldStreamingDecoder overrides mark to ensure sufficient buffer size, but this assumption may not hold for all input streams passed to this method. Consider checking if mark is supported using markSupported() before calling mark, or document that the input stream must support mark/reset.

Suggested change

if (methodDescriptor == null) {

if (methodDescriptor == null) {

if (!rawMessage.markSupported()) {

throw new IOException("InputStream must support mark/reset to resolve overloaded triple methods");

}

...tp12/src/main/java/org/apache/dubbo/remoting/http12/message/LengthFieldStreamingDecoder.java

...riple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/AbstractTripleClientStream.java

…protocol/tri/stream/AbstractTripleClientStream.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

…ces for InputStream

This commit unifies the deframer implementation between client and server sides for the Triple protocol, eliminating code duplication and improving maintainability. ## Key Changes ### Deleted Files - TriDecoder.java: Removed client-specific deframer implementation - Deframer.java: Removed client deframer interface - RecordListener.java: Removed test helper - TriDecoderTest.java: Removed corresponding test ### Core Modifications 1. **LengthFieldStreamingDecoder**: Added BoundedInputStream to support: - Message boundary isolation (prevents reading beyond current message) - mark/reset support required by Hessian2 and other deserializers - Zero-copy optimization by reading directly from accumulate stream 2. **StreamingDecoder.FragmentListener**: Added messageLength parameter to onFragmentMessage() method for better flow control 3. **AbstractTripleClientStream**: Migrated from TriDecoder to GrpcStreamingDecoder, using ByteBufInputStream to adapt Netty's ByteBuf to InputStream 4. **Stream.Listener.onMessage**: Changed parameter from byte[] to InputStream for unified handling and memory optimization 5. **PackableMethod**: Added default parseResponse(InputStream) method for backward compatibility 6. **CompositeInputStream**: Improved stream release logic to prevent exceptions 7. **DescriptorUtils**: Added mark() call before reading stream to support reset ## Architecture Change Before: - Client: ByteBuf -> TriDecoder (Netty-specific) -> byte[] -> deserializer - Server: InputStream -> LengthFieldStreamingDecoder -> byte[] -> deserializer After: - Client: ByteBuf -> ByteBufInputStream -> GrpcStreamingDecoder -> BoundedInputStream -> deserializer - Server: InputStream -> GrpcStreamingDecoder -> BoundedInputStream -> deserializer ## Bug Fix Fixed BoundedInputStream.reset() not restoring the 'remaining' counter, which caused streams to return EOF after mark/reset. This was the root cause of POJO method invocation failures with "Unexpected serialization type:null" error.

…ces for InputStream

…putStream

# Conflicts: # dubbo-common/src/main/java/org/apache/dubbo/rpc/model/PackableMethod.java # dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/DescriptorUtils.java # dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/stream/AbstractTripleClientStream.java

…putStream

# Conflicts: # dubbo-common/src/main/java/org/apache/dubbo/rpc/model/PackableMethod.java # dubbo-metrics/dubbo-metrics-otlp/src/main/java/org/apache/dubbo/metrics/otlp/OtlpMetricsReporterFactory.java # dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/DescriptorUtils.java

zrlw

LGTM

RainYuY · 2026-02-05T03:51:58Z

...pc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleCustomerProtocolWrapper.java

+                if ((b & 0x80) != 0) {
+                    int shift = 7;
+                    tag = b & 0x7F;
+                    while (shift < 35) {


There is duplicate code here—please extract it into a separate method.

We will remove the byte-based methods in subsequent versions.

RainYuY · 2026-02-05T03:53:21Z

...pc-triple/src/main/java/org/apache/dubbo/rpc/protocol/tri/TripleCustomerProtocolWrapper.java

+                        }
+                        tag |= (b & 0x7F) << shift;
+                        if ((b & 0x80) == 0) {
+                            break;


Have we encoded it using this method?

To avoid relying on the Protobuf (PB) dependency, we implemented the PB encoding structure directly ourselves.

EarthChen and others added 16 commits January 20, 2026 18:47

refactor: enhance backpressure handling with byte tracking in HTTP/2 …

4bf68fb

…streams

Update dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/…

fc0718a

…dubbo/remoting/http12/h2/Http2ServerChannelObserver.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Update dubbo-remoting/dubbo-remoting-http12/src/main/java/org/apache/…

3272a1d

…dubbo/remoting/http12/h2/Http2ServerChannelObserver.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

fix

13ed0fe

refactor: expose methods for byte tracking in AbstractTripleClientStr…

6d7d1ba

…eam and Http2ServerChannelObserver

refactor: enhance error handling and byte tracking in HTTP/2 stream o…

244ba4e

…bservers

refactor: improve closing logic and state management in LengthFieldSt…

41a031b

…reamingDecoder

Merge branch '3.3' of github.com:apache/dubbo into 3.3

05deae1

fix: prevent race condition in stream initialization

f7de211

Merge branch 'apache:3.3' into 3.3

3c025c9

Merge branch '3.3' of github.com:EarthChen/dubbo into 3.3

abb678d

Merge branch 'fix-init-stream-race' into 3.3

96714b8

refactor: update tri serialization methods to use InputStream and Out…

ae2600c

…putStream

Merge branch '3.3' of github.com:apache/dubbo into refactor-tri-drframer

a9d1e39

refactor: update tri serialization methods to use InputStream and Out…

52cabcf

…putStream

EarthChen changed the title ~~Refactor packable stream~~ Refactor tri pack & unpack use stream Jan 23, 2026

Merge branch '3.3' into refactor-packable-stream

07caa6b

zrlw requested a review from Copilot January 24, 2026 11:31

Copilot started reviewing on behalf of zrlw January 24, 2026 11:31 View session

Copilot AI reviewed Jan 24, 2026

View reviewed changes

EarthChen and others added 8 commits January 24, 2026 20:41

Update dubbo-rpc/dubbo-rpc-triple/src/main/java/org/apache/dubbo/rpc/…

dbe9d7d

…protocol/tri/stream/AbstractTripleClientStream.java Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Merge branch 'refactor-tri-drframer' into refactor-packable-stream

598a2a7

refactor: enhance error handling in deframer by using try-with-resour…

3128bd2

…ces for InputStream

fix

4ba46e6

refactor: enhance error handling in deframer by using try-with-resour…

aed26d0

…ces for InputStream

refactor: update tri serialization methods to use InputStream and Out…

c97786e

…putStream

fix

ccf7d0d

EarthChen added 4 commits January 24, 2026 22:46

refactor: update tri serialization methods to use InputStream and Out…

9f2f062

…putStream

Merge branch 'refactor-tri-drframer' into refactor-packable-stream

9f88aac

fix

7a35d21

EarthChen requested a review from RainYuY February 3, 2026 02:47

EarthChen added 4 commits February 4, 2026 10:38

Merge branch '3.3' of github.com:EarthChen/dubbo into 3.3

3961b0c

Merge branch '3.3' into refactor-packable-stream

a341043

Merge branch '3.3' of github.com:apache/dubbo into 3.3

a319765

zrlw approved these changes Feb 4, 2026

View reviewed changes

zrlw added the status/wait for another approve label Feb 4, 2026

RainYuY requested changes Feb 5, 2026

View reviewed changes

RainYuY approved these changes Feb 5, 2026

View reviewed changes

EarthChen merged commit 678d2a7 into apache:3.3 Feb 5, 2026
33 checks passed

-                int tag = b;
-                if ((b & 0x80) != 0) {
-                    int shift = 7;
-                    tag = b & 0x7F;
-                    while (shift < 35) {
-                        b = inputStream.read();
-                        if (b == -1) {
-                            throw new IOException("Unexpected end of stream while reading tag");
-                        }
-                        tag |= (b & 0x7F) << shift;
-                        if ((b & 0x80) == 0) {
-                            break;
-                        }
-                        shift += 7;
-                    }
-                }
+                final int firstByte = b;
+                InputStream tagInputStream = new InputStream() {
+                    private boolean first = true;
+                    @Override
+                    public int read() throws IOException {
+                        if (first) {
+                            first = false;
+                            return firstByte;
+                        }
+                        return inputStream.read();
+                    }
+                    @Override
+                    public int read(byte[] buffer, int off, int len) throws IOException {
+                        if (buffer == null) {
+                            throw new NullPointerException();
+                        }
+                        if (off < 0 || len < 0 || len > buffer.length - off) {
+                            throw new IndexOutOfBoundsException();
+                        }
+                        if (len == 0) {
+                            return 0;
+                        }
+                        if (first) {
+                            first = false;
+                            buffer[off] = (byte) firstByte;
+                            return 1;
+                        }
+                        return inputStream.read(buffer, off, len);
+                    }
+                };
+                int tag = readRawVarint32(tagInputStream);

Conversation

EarthChen commented Jan 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

What is the purpose of the change?

Checklist

Uh oh!

codecov-commenter commented Jan 23, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Copilot AI Jan 24, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 24, 2026

Choose a reason for hiding this comment

Uh oh!

Copilot AI Jan 24, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Copilot AI Jan 24, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

zrlw left a comment

Choose a reason for hiding this comment

Uh oh!

RainYuY Feb 5, 2026

Choose a reason for hiding this comment

Uh oh!

EarthChen Feb 5, 2026

Choose a reason for hiding this comment

Uh oh!

RainYuY Feb 5, 2026

Choose a reason for hiding this comment

Uh oh!

EarthChen Feb 5, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

EarthChen commented Jan 23, 2026 •

edited

Loading

codecov-commenter commented Jan 23, 2026 •

edited

Loading