This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new bb996b4a50 [ISSUE #11283] fix(remoting): RpcClientUtils.encodeBody
must not clobber the caller's ByteBuffer mark (#11296)
bb996b4a50 is described below
commit bb996b4a5053a644375ceee2a8d897df4312aec2
Author: 健康宝 <[email protected]>
AuthorDate: Thu Oct 8 10:46:11 2026 +0800
[ISSUE #11283] fix(remoting): RpcClientUtils.encodeBody must not clobber
the caller's ByteBuffer mark (#11296)
encodeBody used mark()/get()/reset() on the caller's buffer, which
restored the position but silently overwrote any mark the caller had set,
so a later reset() jumped to the wrong position. Read the remaining bytes
from a duplicate() instead, leaving the original buffer's position, limit
and mark untouched, and add a regression test.
Co-authored-by: Claude Opus 5 (1M context) <[email protected]>
---
.../rocketmq/remoting/rpc/RpcClientUtils.java | 4 +-
.../rocketmq/remoting/rpc/RpcClientUtilsTest.java | 87 ++++++++++++++++++++++
2 files changed, 88 insertions(+), 3 deletions(-)
diff --git
a/remoting/src/main/java/org/apache/rocketmq/remoting/rpc/RpcClientUtils.java
b/remoting/src/main/java/org/apache/rocketmq/remoting/rpc/RpcClientUtils.java
index 78a33b72c5..68f16807c2 100644
---
a/remoting/src/main/java/org/apache/rocketmq/remoting/rpc/RpcClientUtils.java
+++
b/remoting/src/main/java/org/apache/rocketmq/remoting/rpc/RpcClientUtils.java
@@ -44,11 +44,9 @@ public class RpcClientUtils {
} else if (body instanceof RemotingSerializable) {
return ((RemotingSerializable) body).encode();
} else if (body instanceof ByteBuffer) {
- ByteBuffer buffer = (ByteBuffer)body;
- buffer.mark();
+ ByteBuffer buffer = ((ByteBuffer)body).duplicate();
byte[] data = new byte[buffer.remaining()];
buffer.get(data);
- buffer.reset();
return data;
} else {
throw new RuntimeException("Unsupported body type " +
body.getClass());
diff --git
a/remoting/src/test/java/org/apache/rocketmq/remoting/rpc/RpcClientUtilsTest.java
b/remoting/src/test/java/org/apache/rocketmq/remoting/rpc/RpcClientUtilsTest.java
new file mode 100644
index 0000000000..1208ca7cd7
--- /dev/null
+++
b/remoting/src/test/java/org/apache/rocketmq/remoting/rpc/RpcClientUtilsTest.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.remoting.rpc;
+
+import java.nio.ByteBuffer;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class RpcClientUtilsTest {
+
+ @Test
+ public void testEncodeBodyByteBufferPreservesCallerMarkAndPosition() {
+ ByteBuffer buffer = ByteBuffer.allocate(5);
+ buffer.put(new byte[] {1, 2, 3, 4, 5});
+ buffer.flip();
+ buffer.position(1);
+ buffer.mark();
+ buffer.position(3);
+
+ byte[] data = RpcClientUtils.encodeBody(buffer);
+
+ assertThat(data).hasSize(2);
+ assertThat(data).containsExactly((byte) 4, (byte) 5);
+ assertThat(buffer.position()).isEqualTo(3);
+ assertThat(buffer.limit()).isEqualTo(5);
+ // Regression for issue #11283: the caller's mark must survive
encodeBody,
+ // so reset() rewinds to the marked position instead of the current
one.
+ buffer.reset();
+ assertThat(buffer.position()).isEqualTo(1);
+ }
+
+ @Test
+ public void testEncodeBodyByteBufferReadsRemainingBytes() {
+ ByteBuffer buffer = ByteBuffer.allocate(5);
+ buffer.put(new byte[] {1, 2, 3, 4, 5});
+ buffer.flip();
+
+ byte[] data = RpcClientUtils.encodeBody(buffer);
+
+ assertThat(data).containsExactly((byte) 1, (byte) 2, (byte) 3, (byte)
4, (byte) 5);
+ assertThat(buffer.position()).isEqualTo(0);
+ assertThat(buffer.limit()).isEqualTo(5);
+ }
+
+ @Test
+ public void testEncodeBodyByteBufferReadOnly() {
+ ByteBuffer buffer = ByteBuffer.allocate(5);
+ buffer.put(new byte[] {1, 2, 3, 4, 5});
+ buffer.flip();
+ buffer.position(2);
+ ByteBuffer readOnlyBuffer = buffer.asReadOnlyBuffer();
+
+ byte[] data = RpcClientUtils.encodeBody(readOnlyBuffer);
+
+ assertThat(data).containsExactly((byte) 3, (byte) 4, (byte) 5);
+ assertThat(readOnlyBuffer.position()).isEqualTo(2);
+ assertThat(readOnlyBuffer.limit()).isEqualTo(5);
+ }
+
+ @Test
+ public void testEncodeBodyNullReturnsNull() {
+ assertThat(RpcClientUtils.encodeBody(null)).isNull();
+ }
+
+ @Test
+ public void testEncodeBodyByteArrayPassthrough() {
+ byte[] body = new byte[] {1, 2, 3};
+
+ assertThat(RpcClientUtils.encodeBody(body)).isSameAs(body);
+ }
+}
