This is an automated email from the ASF dual-hosted git repository. mmodzelewski pushed a commit to branch java-strings-serialization in repository https://gitbox.apache.org/repos/asf/iggy.git
commit c9eb389e329c1779ad3852604886dd43e36329e9 Author: Maciej Modzelewski <[email protected]> AuthorDate: Sat Sep 5 14:26:40 2026 +0200 fix(java): size wire strings by UTF-8 byte length, not char count Non-ASCII names, keys, passwords and login metadata went out with a length prefix taken from String.length(), which counts UTF-16 code units, while the bytes that followed came from getBytes() in the platform default charset. Any character outside ASCII made the prefix disagree with the payload and the server misread the frame. The 255 cap on identifiers and message keys was also checked in characters, so a name under 255 chars but over 255 bytes overflowed the u8 prefix. Every wire string is now encoded as UTF-8 once and prefixed with the length of those bytes. Identifier and Partitioning validate the cap against the encoded length and Identifier.getSize reports it, so buffer sizing matches what is written. Message.of encodes its payload as UTF-8 explicitly instead of relying on the JVM default. Refs #4056 --- .../iggy/client/async/tcp/UsersTcpClient.java | 27 ++++++---- .../org/apache/iggy/identifier/Identifier.java | 13 ++++- .../main/java/org/apache/iggy/message/Message.java | 3 +- .../java/org/apache/iggy/message/Partitioning.java | 16 ++++-- .../org/apache/iggy/serde/BytesSerializer.java | 10 ++-- .../async/tcp/UsersTcpClientPayloadTest.java | 63 ++++++++++++++++++++++ .../client/blocking/tcp/BytesSerializerTest.java | 41 ++++++++++++++ .../org/apache/iggy/identifier/IdentifierTest.java | 24 +++++++++ .../java/org/apache/iggy/message/MessageTest.java | 9 ++++ .../org/apache/iggy/message/PartitioningTest.java | 19 +++++++ 10 files changed, 203 insertions(+), 22 deletions(-) diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java index 1bf1e7ced..6b3d9b454 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java @@ -19,6 +19,7 @@ package org.apache.iggy.client.async.tcp; +import io.netty.buffer.ByteBuf; import io.netty.buffer.Unpooled; import org.apache.iggy.IggyVersion; import org.apache.iggy.client.async.UsersClient; @@ -35,6 +36,7 @@ import org.apache.iggy.user.UserStatus; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; import java.util.Optional; @@ -155,20 +157,23 @@ public class UsersTcpClient implements UsersClient { return routingHook.loginOnLeader(() -> loginWithoutRedirect(username, password)); } + static ByteBuf loginPayload(String username, String password, String version, String context) { + byte[] versionBytes = version.getBytes(StandardCharsets.UTF_8); + byte[] contextBytes = context.getBytes(StandardCharsets.UTF_8); + var payload = Unpooled.buffer(); + payload.writeBytes(toBytes(username)); + payload.writeBytes(toBytes(password)); + payload.writeIntLE(versionBytes.length); + payload.writeBytes(versionBytes); + payload.writeIntLE(contextBytes.length); + payload.writeBytes(contextBytes); + return payload; + } + private CompletableFuture<IdentityInfo> loginWithoutRedirect(String username, String password) { String version = IggyVersion.getInstance().getUserAgent(); String context = IggyVersion.getInstance().toString(); - - var payload = Unpooled.buffer(); - var usernameBytes = toBytes(username); - var passwordBytes = toBytes(password); - - payload.writeBytes(usernameBytes); - payload.writeBytes(passwordBytes); - payload.writeIntLE(version.length()); - payload.writeBytes(version.getBytes()); - payload.writeIntLE(context.length()); - payload.writeBytes(context.getBytes()); + var payload = loginPayload(username, password, version, context); log.debug("Logging in user: {}", username); diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/identifier/Identifier.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/identifier/Identifier.java index 945d35696..5c71a6256 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/identifier/Identifier.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/identifier/Identifier.java @@ -23,9 +23,13 @@ import org.apache.commons.lang3.StringUtils; import org.apache.iggy.exception.IggyInvalidArgumentException; import javax.annotation.Nullable; +import java.nio.charset.StandardCharsets; public abstract class Identifier { + /** Server-side cap on a wire name, in UTF-8 bytes, matching its u8 length prefix. */ + public static final int MAX_NAME_LENGTH = 255; + private final String name; private final Long id; @@ -37,6 +41,11 @@ public abstract class Identifier { throw new IggyInvalidArgumentException("Name and id cannot be both present"); } if (StringUtils.isNotBlank(name)) { + int encodedLength = name.getBytes(StandardCharsets.UTF_8).length; + if (encodedLength > MAX_NAME_LENGTH) { + throw new IggyInvalidArgumentException( + "Name must be at most " + MAX_NAME_LENGTH + " bytes, got " + encodedLength); + } this.name = name; this.id = null; } else { @@ -73,8 +82,8 @@ public abstract class Identifier { // kind, 1 byte + length, 1 byte + id, 4 bytes return 6; } else { - // kind, 1 byte + length, 1 byte + name.length() - return 2 + name.length(); + // kind, 1 byte + length, 1 byte + encoded name bytes + return 2 + name.getBytes(StandardCharsets.UTF_8).length; } } } diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Message.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Message.java index aa5389f26..d7249df15 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Message.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Message.java @@ -21,6 +21,7 @@ package org.apache.iggy.message; import javax.annotation.Nullable; import java.math.BigInteger; +import java.nio.charset.StandardCharsets; import java.util.Collections; import java.util.HashMap; import java.util.List; @@ -46,7 +47,7 @@ public record Message(MessageHeader header, byte[] payload, Map<HeaderKey, Heade } public static Message of(String payload, Map<HeaderKey, HeaderValue> userHeaders) { - final byte[] payloadBytes = payload.getBytes(); + final byte[] payloadBytes = payload.getBytes(StandardCharsets.UTF_8); final long userHeadersLength = getUserHeadersSize(userHeaders); final MessageHeader msgHeader = new MessageHeader( BigInteger.ZERO, diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java index 51f071fc3..91b3a01b8 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java @@ -23,8 +23,13 @@ import org.apache.commons.lang3.ArrayUtils; import org.apache.iggy.exception.IggyInvalidArgumentException; import java.nio.ByteBuffer; +import java.nio.charset.StandardCharsets; public record Partitioning(PartitioningKind kind, byte[] value) { + + /** Server-side cap on a messages key, in encoded bytes, matching its u8 length prefix. */ + public static final int MAX_MESSAGES_KEY_LENGTH = 255; + public static Partitioning balanced() { return new Partitioning(PartitioningKind.Balanced, new byte[] {}); } @@ -38,10 +43,15 @@ public record Partitioning(PartitioningKind kind, byte[] value) { } public static Partitioning messagesKey(String key) { - if (key == null || key.isBlank() || key.length() > 255) { - throw new IggyInvalidArgumentException("Key must be non-empty and less than 255 characters long"); + if (key == null || key.isBlank()) { + throw new IggyInvalidArgumentException("Key must be non-empty"); + } + byte[] encoded = key.getBytes(StandardCharsets.UTF_8); + if (encoded.length > MAX_MESSAGES_KEY_LENGTH) { + throw new IggyInvalidArgumentException( + "Key must be at most " + MAX_MESSAGES_KEY_LENGTH + " bytes, got " + encoded.length); } - return new Partitioning(PartitioningKind.MessagesKey, key.getBytes()); + return new Partitioning(PartitioningKind.MessagesKey, encoded); } public int getSize() { diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java index 6a67b138f..87afdcc7a 100644 --- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java +++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java @@ -86,10 +86,10 @@ public final class BytesSerializer { buffer.writeIntLE(identifier.getId().intValue()); return buffer; } else if (identifier.getKind() == 2) { - ByteBuf buffer = Unpooled.buffer(2 + identifier.getName().length()); + ByteBuf name = toBytes(identifier.getName()); + ByteBuf buffer = Unpooled.buffer(1 + name.readableBytes()); buffer.writeByte(2); - buffer.writeByte(identifier.getName().length()); - buffer.writeBytes(identifier.getName().getBytes()); + buffer.writeBytes(name); return buffer; } else { throw new IggyInvalidArgumentException("Unknown identifier kind: " + identifier.getKind()); @@ -210,9 +210,9 @@ public final class BytesSerializer { } public static ByteBuf toBytes(String value) { - int bufferLength = 1 + value.length(); - ByteBuf buffer = Unpooled.buffer(bufferLength); byte[] stringBytes = value.getBytes(StandardCharsets.UTF_8); + int bufferLength = 1 + stringBytes.length; + ByteBuf buffer = Unpooled.buffer(bufferLength); buffer.writeByte(stringBytes.length); buffer.writeBytes(stringBytes); return buffer; diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/UsersTcpClientPayloadTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/UsersTcpClientPayloadTest.java new file mode 100644 index 000000000..7ee3d237a --- /dev/null +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/UsersTcpClientPayloadTest.java @@ -0,0 +1,63 @@ +/* + * 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.iggy.client.async.tcp; + +import io.netty.buffer.ByteBuf; +import org.junit.jupiter.api.Test; + +import java.nio.charset.StandardCharsets; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * LoginUser wire format: + * [username_len:u8][username:N][password_len:u8][password:N] + * [version_len:u32_le][version:N][context_len:u32_le][context:N] + */ +class UsersTcpClientPayloadTest { + + @Test + void shouldPrefixEveryFieldWithItsUtf8ByteLength() { + String username = "użytkownik"; + String password = "hasło"; + String version = "iggy-java-sdk/0.9.0"; + String context = "build: 2026-09-05, commit: ąęó"; + + ByteBuf payload = UsersTcpClient.loginPayload(username, password, version, context); + + assertThat(readU8String(payload)).isEqualTo(username); + assertThat(readU8String(payload)).isEqualTo(password); + assertThat(readU32String(payload)).isEqualTo(version); + assertThat(readU32String(payload)).isEqualTo(context); + assertThat(payload.isReadable()).isFalse(); + } + + private static String readU8String(ByteBuf buffer) { + byte[] bytes = new byte[buffer.readUnsignedByte()]; + buffer.readBytes(bytes); + return new String(bytes, StandardCharsets.UTF_8); + } + + private static String readU32String(ByteBuf buffer) { + byte[] bytes = new byte[Math.toIntExact(buffer.readUnsignedIntLE())]; + buffer.readBytes(bytes); + return new String(bytes, StandardCharsets.UTF_8); + } +} diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/BytesSerializerTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/BytesSerializerTest.java index 091ffc189..65dda86b0 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/BytesSerializerTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/BytesSerializerTest.java @@ -20,6 +20,7 @@ package org.apache.iggy.client.blocking.tcp; import io.netty.buffer.ByteBuf; +import io.netty.buffer.ByteBufUtil; import io.netty.buffer.Unpooled; import org.apache.iggy.consumergroup.Consumer; import org.apache.iggy.exception.IggyInvalidArgumentException; @@ -36,6 +37,8 @@ import org.apache.iggy.user.StreamPermissions; import org.apache.iggy.user.TopicPermissions; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import java.math.BigInteger; import java.nio.charset.StandardCharsets; @@ -239,6 +242,44 @@ class BytesSerializerTest { result.readBytes(nameBytes); assertThat(new String(nameBytes)).isEqualTo("test-stream"); } + + @Test + void shouldSerializeUtf8StringIdentifierWithByteLength() { + // given + String name = "strumień-世界"; + byte[] expectedBytes = name.getBytes(StandardCharsets.UTF_8); + var identifier = StreamId.of(name); + + // when + ByteBuf result = BytesSerializer.toBytes(identifier); + + // then + assertThat(expectedBytes.length).isGreaterThan(name.length()); + assertThat(result.readableBytes()).isEqualTo(identifier.getSize()); + assertThat(result.readByte()).isEqualTo((byte) 2); // kind = 2 (string) + assertThat(result.readByte()).isEqualTo((byte) expectedBytes.length); + byte[] nameBytes = new byte[expectedBytes.length]; + result.readBytes(nameBytes); + assertThat(nameBytes).isEqualTo(expectedBytes); + assertThat(result.readableBytes()).isEqualTo(0); + } + + @ParameterizedTest + @CsvSource({ + "café, 02 05 63 61 66 C3 A9", + "naïve-café, 02 0C 6E 61 C3 AF 76 65 2D 63 61 66 C3 A9", + "日本語, 02 09 E6 97 A5 E6 9C AC E8 AA 9E", + }) + void shouldMatchServerWireFormatForNonAsciiNames(String name, String expectedHex) { + // given + var identifier = StreamId.of(name); + + // when + ByteBuf result = BytesSerializer.toBytes(identifier); + + // then + assertThat(ByteBufUtil.hexDump(result).toUpperCase()).isEqualTo(expectedHex.replace(" ", "")); + } } @Nested diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/identifier/IdentifierTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/identifier/IdentifierTest.java index cc7bee7a9..a5211becd 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/identifier/IdentifierTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/identifier/IdentifierTest.java @@ -23,6 +23,9 @@ import org.apache.iggy.exception.IggyInvalidArgumentException; import org.jspecify.annotations.Nullable; import org.junit.jupiter.api.Test; +import java.nio.charset.StandardCharsets; + +import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; public class IdentifierTest { @@ -31,6 +34,27 @@ public class IdentifierTest { assertThatThrownBy(() -> new FakeIdentifier("foo", 123L)).isInstanceOf(IggyInvalidArgumentException.class); } + @Test + void getSizeCountsEncodedBytesOfName() { + assertThat(new FakeIdentifier("世界", null).getSize()).isEqualTo(2 + 6); + } + + @Test + void constructorAcceptsNameOfExactly255EncodedBytes() { + String name = "世".repeat(85); + assertThat(name.getBytes(StandardCharsets.UTF_8)).hasSize(255); + assertThat(new FakeIdentifier(name, null).getName()).isEqualTo(name); + } + + @Test + void constructorThrowsWhenNameExceeds255EncodedBytesEvenIfUnder255Chars() { + String name = "あ".repeat(200); + assertThat(name.length()).isLessThan(255); + assertThatThrownBy(() -> new FakeIdentifier(name, null)) + .isInstanceOf(IggyInvalidArgumentException.class) + .hasMessageContaining("600"); + } + static class FakeIdentifier extends Identifier { protected FakeIdentifier(@Nullable String name, @Nullable Long id) { super(name, id); diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/message/MessageTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/message/MessageTest.java index 48cb5a8be..927dcc3ba 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/message/MessageTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/message/MessageTest.java @@ -22,6 +22,7 @@ package org.apache.iggy.message; import org.junit.jupiter.api.Test; import java.math.BigInteger; +import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; @@ -72,6 +73,14 @@ class MessageTest { assertThat(message.userHeaders().size()).isEqualTo(0); } + @Test + void ofEncodesStringPayloadAsUtf8() { + var message = Message.of("世界"); + + assertThat(message.payload()).isEqualTo("世界".getBytes(StandardCharsets.UTF_8)); + assertThat(message.header().payloadLength()).isEqualTo(6L); + } + @Test void ofCreatesExpectedMessageWhenGivenPayloadOnly() { var message = Message.of("foo"); diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/message/PartitioningTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/message/PartitioningTest.java index 3fd35ac32..ab476eaa3 100644 --- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/message/PartitioningTest.java +++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/message/PartitioningTest.java @@ -25,6 +25,8 @@ import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.NullSource; import org.junit.jupiter.params.provider.ValueSource; +import java.nio.charset.StandardCharsets; + import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -59,6 +61,23 @@ class PartitioningTest { assertThatThrownBy(() -> Partitioning.messagesKey(id)).isInstanceOf(IggyInvalidArgumentException.class); } + @Test + void messagesKeyThrowsIggyInvalidArgumentExceptionWhenEncodedValueExceeds255BytesEvenIfUnder255Chars() { + var id = "あ".repeat(100); + assertThat(id.length()).isLessThan(255); + assertThatThrownBy(() -> Partitioning.messagesKey(id)) + .isInstanceOf(IggyInvalidArgumentException.class) + .hasMessageContaining("300"); + } + + @Test + void messagesKeyEncodesValueAsUtf8() { + var result = Partitioning.messagesKey("世界"); + + assertThat(result.value()).isEqualTo("世界".getBytes(StandardCharsets.UTF_8)); + assertThat(result.getSize()).isEqualTo(2 + 6); + } + @Test void messagesKeyReturnsPartitioningWithMessagesKeyKindAndExpectedValueWhenGivenValidArguments() { var id = "the-key";
