You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by ch...@apache.org on 2024/05/06 11:11:10 UTC
(kafka) branch trunk updated: KAFKA-16393 read/write sequence of buffers correctly (#15571)
This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 5c96ad61d95 KAFKA-16393 read/write sequence of buffers correctly (#15571)
5c96ad61d95 is described below
commit 5c96ad61d95245c1333ecdb65562d421a1186c5b
Author: Okada Haruki <oc...@gmail.com>
AuthorDate: Mon May 6 20:11:04 2024 +0900
KAFKA-16393 read/write sequence of buffers correctly (#15571)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../kafka/common/network/SslTransportLayer.java | 4 +-
.../common/network/SslTransportLayerTest.java | 68 ++++++++++++++++++++++
2 files changed, 70 insertions(+), 2 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java b/clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java
index 0b154fb2909..ef7f137e366 100644
--- a/clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java
+++ b/clients/src/main/java/org/apache/kafka/common/network/SslTransportLayer.java
@@ -682,7 +682,7 @@ public class SslTransportLayer implements TransportLayer {
int totalRead = 0;
int i = offset;
- while (i < length) {
+ while (i < offset + length) {
if (dsts[i].hasRemaining()) {
int read = read(dsts[i]);
if (read > 0)
@@ -755,7 +755,7 @@ public class SslTransportLayer implements TransportLayer {
throw new IndexOutOfBoundsException();
int totalWritten = 0;
int i = offset;
- while (i < length) {
+ while (i < offset + length) {
if (srcs[i].hasRemaining() || hasPendingWrites()) {
int written = write(srcs[i]);
if (written > 0) {
diff --git a/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java b/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
index b6033747df3..41264c5cef6 100644
--- a/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/network/SslTransportLayerTest.java
@@ -73,6 +73,7 @@ import javax.net.ssl.SSLException;
import javax.net.ssl.SSLParameters;
import javax.net.ssl.SSLSession;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
@@ -81,7 +82,9 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assumptions.assumeTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoMoreInteractions;
@@ -1558,4 +1561,69 @@ public class SslTransportLayerTest {
verify(sslEngine, times(1)).closeInbound();
verifyNoMoreInteractions(sslEngine);
}
+
+ @Test
+ public void testGatheringWrite() throws IOException {
+ SSLEngine sslEngine = mock(SSLEngine.class);
+ SelectionKey selectionKey = mock(SelectionKey.class);
+ SslTransportLayer sslTransportLayer = spy(new SslTransportLayer(
+ "test-channel",
+ selectionKey,
+ sslEngine,
+ mock(ChannelMetadataRegistry.class)
+ ));
+ doReturn(false).when(sslTransportLayer).hasPendingWrites();
+
+ ByteBuffer mockSocket = ByteBuffer.allocate(1024);
+ when(sslTransportLayer.write(any(ByteBuffer.class))).then(invocation -> {
+ ByteBuffer buf = invocation.getArgument(0);
+ int written = buf.remaining();
+ mockSocket.put(buf);
+ return written;
+ });
+ ByteBuffer[] srcs = {
+ ByteBuffer.wrap("Hello, ".getBytes(StandardCharsets.UTF_8)),
+ ByteBuffer.wrap("World".getBytes(StandardCharsets.UTF_8)),
+ ByteBuffer.wrap("!".getBytes(StandardCharsets.UTF_8))
+ };
+
+ byte[] expected = "World!".getBytes(StandardCharsets.UTF_8);
+ assertEquals(expected.length, sslTransportLayer.write(srcs, 1, 2));
+
+ mockSocket.flip();
+ byte[] actual = new byte[expected.length];
+ mockSocket.get(actual);
+ assertArrayEquals(expected, actual);
+ }
+
+ @Test
+ public void testScatteringRead() throws IOException {
+ SSLEngine sslEngine = mock(SSLEngine.class);
+ SelectionKey selectionKey = mock(SelectionKey.class);
+ SslTransportLayer sslTransportLayer = spy(new SslTransportLayer(
+ "test-channel",
+ selectionKey,
+ sslEngine,
+ mock(ChannelMetadataRegistry.class)
+ ));
+
+ ByteBuffer mockSocket = ByteBuffer.wrap("Hello, World!".getBytes(StandardCharsets.UTF_8));
+ when(sslTransportLayer.read(any(ByteBuffer.class))).then(invocation -> {
+ ByteBuffer buf = invocation.getArgument(0);
+ int read = buf.remaining();
+ for (int i = 0; i < read; i++) {
+ buf.put(mockSocket.get());
+ }
+ return read;
+ });
+ ByteBuffer[] dsts = {
+ ByteBuffer.allocate(2),
+ ByteBuffer.allocate(3),
+ ByteBuffer.allocate(4)
+ };
+
+ assertEquals(7, sslTransportLayer.read(dsts, 1, 2));
+ assertArrayEquals("Hel".getBytes(StandardCharsets.UTF_8), dsts[1].array());
+ assertArrayEquals("lo, ".getBytes(StandardCharsets.UTF_8), dsts[2].array());
+ }
}