This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new b87b6a6f7 fix(java): size wire strings by UTF-8 byte length, not char
count (#4068)
b87b6a6f7 is described below
commit b87b6a6f77dfa650753cc1998f75fc7641f00463
Author: Maciej Modzelewski <[email protected]>
AuthorDate: Mon Sep 7 15:59:49 2026 +0200
fix(java): size wire strings by UTF-8 byte length, not char count (#4068)
Names, keys and passwords with non-ASCII characters went out with a
length prefix counted in UTF-16 chars while the bytes came from the
platform default charset, so the prefix and payload disagreed and the
server misread the frame. The 255 cap was also checked in chars, so a
short name could still overflow the u8 prefix, and uncapped strings
wrapped it silently.
Every wire string is now encoded as UTF-8 once and prefixed with its
byte length. The serializer rejects empty or over-long strings and
names the field. Identifier caches the encoded name, and Partitioning
checks the value shape the server expects for each kind. The SDK
version is cut at a code point boundary. Message.of, the example
consumer and the javadoc all use UTF-8 explicitly.
The login payload no longer carries version and context strings; the
VSR codec drops them and sends the SDK version itself. The codec now
validates fields before allocating, closing a pooled-buffer leak on
an empty username. Usernames, passwords and token names are checked
against the server's own bounds before the round trip.
Message keys are hashed as UTF-8 bytes, so on a JVM whose default
charset is not UTF-8 a non-ASCII key may land on a different
partition than before. Needs a 0.9.0 release note. Non-ASCII names
are tested end to end over TCP only; the HTTP client does not
percent-encode paths yet.
Refs #4056
---
.../apache/iggy/examples/async/AsyncConsumer.java | 3 +-
.../apache/iggy/client/async/MessagesClient.java | 2 +-
.../client/async/tcp/ConsumerGroupsTcpClient.java | 10 +-
.../async/tcp/PersonalAccessTokensTcpClient.java | 14 +-
.../iggy/client/async/tcp/StreamsTcpClient.java | 22 +--
.../iggy/client/async/tcp/TopicsTcpClient.java | 4 +-
.../iggy/client/async/tcp/UsersTcpClient.java | 46 +++---
.../iggy/client/async/tcp/vsr/VsrLoginCodec.java | 58 ++++++--
.../org/apache/iggy/identifier/Identifier.java | 32 ++++-
.../main/java/org/apache/iggy/message/Message.java | 3 +-
.../java/org/apache/iggy/message/Partitioning.java | 42 +++++-
.../org/apache/iggy/serde/BytesSerializer.java | 37 ++---
.../client/async/tcp/vsr/VsrLoginCodecTest.java | 159 +++++++++++++++++++++
.../client/blocking/tcp/BytesSerializerTest.java | 109 ++++++++++++--
.../client/blocking/tcp/MessagesTcpClientTest.java | 35 +++++
.../client/blocking/tcp/StreamTcpClientTest.java | 24 ++++
.../org/apache/iggy/identifier/IdentifierTest.java | 49 +++++++
.../java/org/apache/iggy/message/MessageTest.java | 9 ++
.../org/apache/iggy/message/PartitioningTest.java | 60 ++++++++
19 files changed, 616 insertions(+), 102 deletions(-)
diff --git
a/examples/java/src/main/java/org/apache/iggy/examples/async/AsyncConsumer.java
b/examples/java/src/main/java/org/apache/iggy/examples/async/AsyncConsumer.java
index 99e59c4d2..885a18186 100644
---
a/examples/java/src/main/java/org/apache/iggy/examples/async/AsyncConsumer.java
+++
b/examples/java/src/main/java/org/apache/iggy/examples/async/AsyncConsumer.java
@@ -31,6 +31,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
@@ -312,7 +313,7 @@ public final class AsyncConsumer {
int messageCount = polled.messages().size();
for (Message message : polled.messages()) {
- String payload = new String(message.payload());
+ String payload = new String(message.payload(),
StandardCharsets.UTF_8);
// Simulate message processing (in real app: parse,
validate, store, etc.)
// This could be CPU-intensive or involve blocking I/O
(database, HTTP calls)
diff --git
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/MessagesClient.java
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/MessagesClient.java
index f3b9ad115..7c69be7c6 100644
---
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/MessagesClient.java
+++
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/MessagesClient.java
@@ -59,7 +59,7 @@ import java.util.concurrent.CompletableFuture;
* Consumer.of(1L), PollingStrategy.first(), 100L, true)
* .thenAccept(polled -> {
* for (var msg : polled.messages()) {
- * System.out.println(new String(msg.payload()));
+ * System.out.println(new String(msg.payload(),
StandardCharsets.UTF_8));
* }
* });
* }</pre>
diff --git
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java
index 786bcc766..d763335eb 100644
---
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java
+++
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java
@@ -104,13 +104,9 @@ public class ConsumerGroupsTcpClient implements
ConsumerGroupsClient {
@Override
public CompletableFuture<ConsumerGroupDetails> createConsumerGroup(
StreamId streamId, TopicId topicId, String name) {
- var streamIdBytes = BytesSerializer.toBytes(streamId);
- var topicIdBytes = BytesSerializer.toBytes(topicId);
- var payload = Unpooled.buffer(1 + streamIdBytes.readableBytes() +
topicIdBytes.readableBytes() + name.length());
-
- payload.writeBytes(streamIdBytes);
- payload.writeBytes(topicIdBytes);
- payload.writeBytes(BytesSerializer.toBytes(name));
+ var payload = BytesSerializer.toBytes(streamId);
+ payload.writeBytes(BytesSerializer.toBytes(topicId));
+ payload.writeBytes(BytesSerializer.toBytes(name, "name"));
log.debug("Creating consumer group - Stream: {}, Topic: {}, Name: {}",
streamId, topicId, name);
diff --git
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PersonalAccessTokensTcpClient.java
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PersonalAccessTokensTcpClient.java
index 680384379..37671a873 100644
---
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PersonalAccessTokensTcpClient.java
+++
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PersonalAccessTokensTcpClient.java
@@ -44,6 +44,14 @@ import java.util.function.Supplier;
public class PersonalAccessTokensTcpClient implements
PersonalAccessTokensClient {
private static final Logger log =
LoggerFactory.getLogger(PersonalAccessTokensTcpClient.class);
+ /**
+ * Token name bounds the server enforces, in UTF-8 bytes. Checked here so
a bad name fails
+ * before the round trip instead of as an opaque server error.
+ */
+ private static final int MIN_NAME_LENGTH = 3;
+
+ private static final int MAX_NAME_LENGTH = 30;
+
private final Supplier<AsyncTcpConnection> connectionSupplier;
private final LoginRoutingHook routingHook;
@@ -63,7 +71,7 @@ public class PersonalAccessTokensTcpClient implements
PersonalAccessTokensClient
@Override
public CompletableFuture<RawPersonalAccessToken>
createPersonalAccessToken(String name, BigInteger expiry) {
var payload = Unpooled.buffer();
- payload.writeBytes(BytesSerializer.toBytes(name));
+ payload.writeBytes(BytesSerializer.toBytes(name, "name",
MIN_NAME_LENGTH, MAX_NAME_LENGTH));
payload.writeBytes(BytesSerializer.toBytesAsU64(expiry));
log.debug("Creating personal access token: {}", name);
@@ -102,7 +110,7 @@ public class PersonalAccessTokensTcpClient implements
PersonalAccessTokensClient
@Override
public CompletableFuture<Void> deletePersonalAccessToken(String name) {
- var payload = BytesSerializer.toBytes(name);
+ var payload = BytesSerializer.toBytes(name, "name", MIN_NAME_LENGTH,
MAX_NAME_LENGTH);
log.debug("Deleting personal access token: {}", name);
@@ -119,7 +127,7 @@ public class PersonalAccessTokensTcpClient implements
PersonalAccessTokensClient
}
private CompletableFuture<IdentityInfo> loginWithoutRedirect(String token)
{
- var payload = BytesSerializer.toBytes(token);
+ var payload = BytesSerializer.toBytes(token, "token");
log.debug("Logging in with personal access token");
diff --git
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java
index 77957a5b4..ec246ef41 100644
---
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java
+++
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java
@@ -23,8 +23,6 @@ import io.netty.buffer.Unpooled;
import io.netty.util.ReferenceCounted;
import org.apache.iggy.client.async.StreamsClient;
import org.apache.iggy.identifier.StreamId;
-import org.apache.iggy.message.HeaderKey;
-import org.apache.iggy.message.HeaderValue;
import org.apache.iggy.serde.BytesSerializer;
import org.apache.iggy.serde.CommandCode;
import org.apache.iggy.stream.StreamBase;
@@ -32,7 +30,6 @@ import org.apache.iggy.stream.StreamDetails;
import java.util.ArrayList;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
@@ -58,10 +55,7 @@ public class StreamsTcpClient implements StreamsClient {
@Override
public CompletableFuture<StreamDetails> createStream(String name) {
- var payloadSize = 1 + name.length();
- var payload = Unpooled.buffer(payloadSize);
-
- payload.writeBytes(BytesSerializer.toBytes(name));
+ var payload = BytesSerializer.toBytes(name, "name");
return connection().send(CommandCode.Stream.CREATE.getValue(),
payload).thenApply(response -> {
StreamDetails details = readStreamDetails(response);
@@ -102,15 +96,11 @@ public class StreamsTcpClient implements StreamsClient {
@Override
public CompletableFuture<Void> updateStream(StreamId streamId, String
name) {
- var payloadSize = 1 + name.length();
- var idBytes = toBytes(streamId);
- var payload = Unpooled.buffer(payloadSize + idBytes.capacity());
-
- payload.writeBytes(idBytes);
- payload.writeBytes(BytesSerializer.toBytes(name));
- // Trailing options block. Streams have no catalog keys yet, so the
- // server rejects every key; the empty block is the extension point.
- payload.writeBytes(BytesSerializer.toBytes(Map.<HeaderKey,
HeaderValue>of()));
+ var payload = toBytes(streamId);
+ payload.writeBytes(BytesSerializer.toBytes(name, "name"));
+ // No trailing options block: streams have no catalog keys yet and the
+ // server reads an absent block as empty. Settings will ride one here,
+ // as topics do.
return connection().send(CommandCode.Stream.UPDATE.getValue(),
payload).thenAccept(ReferenceCounted::release);
}
diff --git
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/TopicsTcpClient.java
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/TopicsTcpClient.java
index ff06d86b9..a25371ae4 100644
---
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/TopicsTcpClient.java
+++
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/TopicsTcpClient.java
@@ -134,7 +134,7 @@ public class TopicsTcpClient implements TopicsClient {
var payload = Unpooled.buffer();
payload.writeBytes(toBytes(streamId));
payload.writeIntLE(partitionsCount.intValue());
- payload.writeBytes(BytesSerializer.toBytes(name));
+ payload.writeBytes(BytesSerializer.toBytes(name, "name"));
payload.writeBytes(BytesSerializer.toBytes(
createTopicOptions(compressionAlgorithm, messageExpiry,
maxTopicSize, options)));
return payload;
@@ -181,7 +181,7 @@ public class TopicsTcpClient implements TopicsClient {
var payload = Unpooled.buffer();
payload.writeBytes(toBytes(streamId));
payload.writeBytes(toBytes(topicId));
- payload.writeBytes(BytesSerializer.toBytes(name));
+ payload.writeBytes(BytesSerializer.toBytes(name, "name"));
// Settings ride the options block. A default value means the caller
did
// not set the key, so it is omitted and the server leaves the topic's
// current value alone.
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..6a4979972 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
@@ -20,11 +20,8 @@
package org.apache.iggy.client.async.tcp;
import io.netty.buffer.Unpooled;
-import org.apache.iggy.IggyVersion;
import org.apache.iggy.client.async.UsersClient;
import org.apache.iggy.identifier.UserId;
-import org.apache.iggy.message.HeaderKey;
-import org.apache.iggy.message.HeaderValue;
import org.apache.iggy.serde.BytesDeserializer;
import org.apache.iggy.serde.CommandCode;
import org.apache.iggy.user.IdentityInfo;
@@ -36,7 +33,6 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.util.List;
-import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.function.Supplier;
@@ -50,6 +46,16 @@ import static org.apache.iggy.serde.BytesSerializer.toBytes;
public class UsersTcpClient implements UsersClient {
private static final Logger log =
LoggerFactory.getLogger(UsersTcpClient.class);
+ /**
+ * Credential bounds the server enforces, in UTF-8 bytes. Checked here so
a bad value fails
+ * before the round trip instead of as an opaque server error.
+ */
+ private static final int MIN_USERNAME_LENGTH = 3;
+
+ private static final int MAX_USERNAME_LENGTH = 50;
+ private static final int MIN_PASSWORD_LENGTH = 3;
+ private static final int MAX_PASSWORD_LENGTH = 100;
+
private final Supplier<AsyncTcpConnection> connectionSupplier;
private final LoginRoutingHook routingHook;
@@ -82,8 +88,8 @@ public class UsersTcpClient implements UsersClient {
public CompletableFuture<UserInfoDetails> createUser(
String username, String password, UserStatus status,
Optional<Permissions> permissions) {
var payload = Unpooled.buffer();
- payload.writeBytes(toBytes(username));
- payload.writeBytes(toBytes(password));
+ payload.writeBytes(toBytes(username, "username", MIN_USERNAME_LENGTH,
MAX_USERNAME_LENGTH));
+ payload.writeBytes(toBytes(password, "password", MIN_PASSWORD_LENGTH,
MAX_PASSWORD_LENGTH));
payload.writeByte(status.asCode());
permissions.ifPresentOrElse(
perms -> {
@@ -109,7 +115,7 @@ public class UsersTcpClient implements UsersClient {
username.ifPresentOrElse(
un -> {
payload.writeByte(1);
- payload.writeBytes(toBytes(un));
+ payload.writeBytes(toBytes(un, "username",
MIN_USERNAME_LENGTH, MAX_USERNAME_LENGTH));
},
() -> payload.writeByte(0));
status.ifPresentOrElse(
@@ -118,9 +124,9 @@ public class UsersTcpClient implements UsersClient {
payload.writeByte(s.asCode());
},
() -> payload.writeByte(0));
- // Trailing options block. Users have no catalog keys yet, so the
- // server rejects every key; the empty block is the extension point.
- payload.writeBytes(toBytes(Map.<HeaderKey, HeaderValue>of()));
+ // No trailing options block: users have no catalog keys yet and the
+ // server reads an absent block as empty. Settings will ride one here,
+ // as topics do.
return connection().sendAndRelease(CommandCode.User.UPDATE, payload);
}
@@ -144,8 +150,8 @@ public class UsersTcpClient implements UsersClient {
@Override
public CompletableFuture<Void> changePassword(UserId userId, String
currentPassword, String newPassword) {
var payload = toBytes(userId);
- payload.writeBytes(toBytes(currentPassword));
- payload.writeBytes(toBytes(newPassword));
+ payload.writeBytes(toBytes(currentPassword, "current password",
MIN_PASSWORD_LENGTH, MAX_PASSWORD_LENGTH));
+ payload.writeBytes(toBytes(newPassword, "new password",
MIN_PASSWORD_LENGTH, MAX_PASSWORD_LENGTH));
return connection().sendAndRelease(CommandCode.User.CHANGE_PASSWORD,
payload);
}
@@ -156,19 +162,11 @@ public class UsersTcpClient implements UsersClient {
}
private CompletableFuture<IdentityInfo> loginWithoutRedirect(String
username, String password) {
- String version = IggyVersion.getInstance().getUserAgent();
- String context = IggyVersion.getInstance().toString();
-
+ // The VSR codec re-frames this into a Register and carries the SDK
+ // version itself, so the payload is only the two credentials.
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());
+ payload.writeBytes(toBytes(username, "username", MIN_USERNAME_LENGTH,
MAX_USERNAME_LENGTH));
+ payload.writeBytes(toBytes(password, "password", MIN_PASSWORD_LENGTH,
MAX_PASSWORD_LENGTH));
log.debug("Logging in user: {}", username);
diff --git
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java
index 36c856c3a..5a6e9a1ab 100644
---
a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java
+++
b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java
@@ -25,6 +25,7 @@ import org.apache.iggy.IggyVersion;
import org.apache.iggy.exception.IggyInvalidArgumentException;
import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
/**
* Rewrites serialized login payloads into the
@@ -46,18 +47,27 @@ final class VsrLoginCodec {
static final String SDK_NAME = "java-sdk";
+ /** Bound on a u8-length-prefixed field, in encoded bytes. */
+ static final int MAX_SHORT_FIELD_LENGTH = 255;
+
+ private static final String UNKNOWN_SDK_VERSION = "unknown";
+
+ private static final byte[] SDK_NAME_FIELD =
SDK_NAME.getBytes(StandardCharsets.UTF_8);
+ private static final byte[] SDK_VERSION_FIELD =
+ sdkVersionField(IggyVersion.getInstance().getVersion());
+
private VsrLoginCodec() {}
/**
* {@code LoginUser} (code 38) payload in:
- * {@code
[username:u8-len][password:u8-len][version:u32-len][context:u32-len]}.
- * The trailing version/context strings are superseded by the
- * {@code ClientVersionInfo} prefix and dropped.
+ * {@code [username:u8-len][password:u8-len]}. Anything after the password
+ * is ignored.
*/
static ByteBuf rewriteUserLogin(ByteBufAllocator alloc, ByteBuf
loginPayload) {
ByteBuf in = loginPayload.slice();
byte[] username = readShortField(in, "username");
byte[] password = readShortField(in, "password");
+ requireShortField(username, "username");
ByteBuf body = alloc.buffer();
writeVersionInfo(body);
@@ -75,6 +85,7 @@ final class VsrLoginCodec {
static ByteBuf rewritePatLogin(ByteBufAllocator alloc, ByteBuf
loginPayload) {
ByteBuf in = loginPayload.slice();
byte[] token = readShortField(in, "token");
+ requireShortField(token, "token");
ByteBuf body = alloc.buffer();
writeVersionInfo(body);
@@ -93,16 +104,29 @@ final class VsrLoginCodec {
private static void writeVersionInfo(ByteBuf body) {
body.writeIntLE(PROTOCOL_VERSION);
- writeShortField(body, SDK_NAME.getBytes(StandardCharsets.UTF_8));
- writeShortField(body, sdkVersion().getBytes(StandardCharsets.UTF_8));
+ writeShortField(body, SDK_NAME_FIELD);
+ writeShortField(body, SDK_VERSION_FIELD);
}
- private static String sdkVersion() {
- String version = IggyVersion.getInstance().getVersion();
- if (version == null || version.isEmpty()) {
- return "unknown";
+ /**
+ * An over-long version is cut on the encoded bytes at a code point
boundary, so the field
+ * always fits its u8 prefix and still decodes as UTF-8 on the server.
+ */
+ static byte[] sdkVersionField(String version) {
+ String value = version == null || version.isEmpty() ?
UNKNOWN_SDK_VERSION : version;
+ byte[] encoded = value.getBytes(StandardCharsets.UTF_8);
+ if (encoded.length <= MAX_SHORT_FIELD_LENGTH) {
+ return encoded;
+ }
+ int end = MAX_SHORT_FIELD_LENGTH;
+ while (isContinuationByte(encoded[end])) {
+ end--;
}
- return version.length() > 255 ? version.substring(0, 255) : version;
+ return Arrays.copyOf(encoded, end);
+ }
+
+ private static boolean isContinuationByte(byte value) {
+ return (value & 0xC0) == 0x80;
}
private static byte[] readShortField(ByteBuf in, String field) {
@@ -118,10 +142,18 @@ final class VsrLoginCodec {
return value;
}
- private static void writeShortField(ByteBuf out, byte[] value) {
- if (value.length == 0 || value.length > 255) {
- throw new IggyInvalidArgumentException("Wire name fields must be
1..255 bytes, got " + value.length);
+ /**
+ * Runs before {@code alloc.buffer()}: the encoder releases the body only
once the codec
+ * returns, so a throw after allocation would leak the pooled buffer.
+ */
+ private static void requireShortField(byte[] value, String field) {
+ if (value.length == 0 || value.length > MAX_SHORT_FIELD_LENGTH) {
+ throw new IggyInvalidArgumentException(
+ "Login payload " + field + " must be 1.." +
MAX_SHORT_FIELD_LENGTH + " bytes, got " + value.length);
}
+ }
+
+ private static void writeShortField(ByteBuf out, byte[] value) {
out.writeByte(value.length);
out.writeBytes(value);
}
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..ddf94b0f7 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
@@ -19,14 +19,21 @@
package org.apache.iggy.identifier;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
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. */
+ private static final int MAX_NAME_LENGTH = 255;
+
private final String name;
+ private final byte[] encodedName;
private final Long id;
protected Identifier(@Nullable String name, @Nullable Long id) {
@@ -37,10 +44,17 @@ public abstract class Identifier {
throw new IggyInvalidArgumentException("Name and id cannot be both
present");
}
if (StringUtils.isNotBlank(name)) {
+ byte[] encoded = name.getBytes(StandardCharsets.UTF_8);
+ if (encoded.length > MAX_NAME_LENGTH) {
+ throw new IggyInvalidArgumentException(
+ "Name must be at most " + MAX_NAME_LENGTH + " bytes,
got " + encoded.length);
+ }
this.name = name;
+ this.encodedName = encoded;
this.id = null;
} else {
this.name = null;
+ this.encodedName = null;
this.id = id;
}
}
@@ -68,13 +82,27 @@ public abstract class Identifier {
return name;
}
+ /** Wire encoding: kind, u8 length, then the u32 little-endian id or the
UTF-8 name. */
+ public ByteBuf toBytes() {
+ ByteBuf buffer = Unpooled.buffer(getSize());
+ buffer.writeByte(getKind());
+ if (id != null) {
+ buffer.writeByte(4);
+ buffer.writeIntLE(id.intValue());
+ } else {
+ buffer.writeByte(encodedName.length);
+ buffer.writeBytes(encodedName);
+ }
+ return buffer;
+ }
+
public int getSize() {
if (id != null) {
// 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 + encodedName.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..93e979239 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,14 +23,40 @@ 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) {
+
+ private static final int PARTITION_ID_LENGTH = 4;
+
+ /** Server-side cap on a messages key, in encoded bytes, matching its u8
length prefix. */
+ private static final int MAX_MESSAGES_KEY_LENGTH = 255;
+
+ public Partitioning {
+ if (kind == null || value == null) {
+ throw new IggyInvalidArgumentException("Partitioning kind and
value cannot be null");
+ }
+ boolean valid =
+ switch (kind) {
+ case Balanced -> value.length == 0;
+ case PartitionId -> value.length == PARTITION_ID_LENGTH;
+ case MessagesKey -> value.length >= 1 && value.length <=
MAX_MESSAGES_KEY_LENGTH;
+ };
+ if (!valid) {
+ throw new IggyInvalidArgumentException(
+ kind + " partitioning value must be " +
expectedLength(kind) + " bytes, got " + value.length);
+ }
+ }
+
public static Partitioning balanced() {
return new Partitioning(PartitioningKind.Balanced, new byte[] {});
}
public static Partitioning partitionId(Long id) {
- ByteBuffer buffer = ByteBuffer.allocate(4);
+ if (id == null) {
+ throw new IggyInvalidArgumentException("Partition id cannot be
null");
+ }
+ ByteBuffer buffer = ByteBuffer.allocate(PARTITION_ID_LENGTH);
buffer.putInt(id.intValue());
byte[] partitionId = buffer.array();
ArrayUtils.reverse(partitionId);
@@ -38,14 +64,22 @@ 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");
}
- return new Partitioning(PartitioningKind.MessagesKey, key.getBytes());
+ return new Partitioning(PartitioningKind.MessagesKey,
key.getBytes(StandardCharsets.UTF_8));
}
public int getSize() {
// kind, 1 byte + length, 1 byte + value.length()
return 2 + value.length;
}
+
+ private static String expectedLength(PartitioningKind kind) {
+ return switch (kind) {
+ case Balanced -> "0";
+ case PartitionId -> String.valueOf(PARTITION_ID_LENGTH);
+ case MessagesKey -> "1.." + MAX_MESSAGES_KEY_LENGTH;
+ };
+ }
}
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 7a20e27fa..51dcaed93 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
@@ -61,6 +61,9 @@ public final class BytesSerializer {
*/
private static final int MAX_HEADER_FIELD_LENGTH = 255;
+ /** Bound on a u8-length-prefixed wire string, in encoded bytes. */
+ private static final int MAX_U8_STRING_LENGTH = 255;
+
/** The timestamp delta is a u32 microsecond offset from the batch origin
timestamp. */
private static final BigInteger MAX_TIMESTAMP_DELTA_MICROS =
BigInteger.valueOf(0xFFFF_FFFFL);
@@ -77,21 +80,7 @@ public final class BytesSerializer {
}
public static ByteBuf toBytes(Identifier identifier) {
- if (identifier.getKind() == 1) {
- ByteBuf buffer = Unpooled.buffer(6);
- buffer.writeByte(1);
- buffer.writeByte(4);
- buffer.writeIntLE(identifier.getId().intValue());
- return buffer;
- } else if (identifier.getKind() == 2) {
- ByteBuf buffer = Unpooled.buffer(2 +
identifier.getName().length());
- buffer.writeByte(2);
- buffer.writeByte(identifier.getName().length());
- buffer.writeBytes(identifier.getName().getBytes());
- return buffer;
- } else {
- throw new IggyInvalidArgumentException("Unknown identifier kind: "
+ identifier.getKind());
- }
+ return identifier.toBytes();
}
public static ByteBuf toBytes(Partitioning partitioning) {
@@ -207,10 +196,22 @@ public final class BytesSerializer {
return buffer;
}
- public static ByteBuf toBytes(String value) {
- int bufferLength = 1 + value.length();
- ByteBuf buffer = Unpooled.buffer(bufferLength);
+ /** A u8-length-prefixed wire string; {@code field} names it in the error
when it does not fit. */
+ public static ByteBuf toBytes(String value, String field) {
+ return toBytes(value, field, 1, MAX_U8_STRING_LENGTH);
+ }
+
+ /**
+ * A u8-length-prefixed wire string bounded to {@code [minLength,
maxLength]} UTF-8 bytes, for
+ * fields the server holds to a tighter range than the prefix allows.
+ */
+ public static ByteBuf toBytes(String value, String field, int minLength,
int maxLength) {
byte[] stringBytes = value.getBytes(StandardCharsets.UTF_8);
+ if (stringBytes.length < minLength || stringBytes.length > maxLength) {
+ throw new IggyInvalidArgumentException("Invalid " + field + "
length: " + stringBytes.length
+ + " bytes when UTF-8 encoded, must be between " +
minLength + " and " + maxLength);
+ }
+ ByteBuf buffer = Unpooled.buffer(1 + stringBytes.length);
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/vsr/VsrLoginCodecTest.java
b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodecTest.java
new file mode 100644
index 000000000..085da287d
--- /dev/null
+++
b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodecTest.java
@@ -0,0 +1,159 @@
+/*
+ * 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.vsr;
+
+import io.netty.buffer.AbstractByteBufAllocator;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.buffer.UnpooledByteBufAllocator;
+import org.apache.iggy.exception.IggyInvalidArgumentException;
+import org.apache.iggy.serde.BytesSerializer;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EmptySource;
+import org.junit.jupiter.params.provider.NullSource;
+
+import java.nio.charset.StandardCharsets;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class VsrLoginCodecTest {
+
+ @Test
+ void sdkVersionFieldEncodesVersionAsUtf8() {
+ assertThat(VsrLoginCodec.sdkVersionField("0.9.0-SNAPSHOT"))
+ .isEqualTo("0.9.0-SNAPSHOT".getBytes(StandardCharsets.UTF_8));
+ }
+
+ @ParameterizedTest
+ @NullSource
+ @EmptySource
+ void sdkVersionFieldFallsBackToUnknownWhenVersionIsMissing(String version)
{
+
assertThat(VsrLoginCodec.sdkVersionField(version)).isEqualTo("unknown".getBytes(StandardCharsets.UTF_8));
+ }
+
+ @Test
+ void sdkVersionFieldTruncatesOnCodePointBoundaryWithinU8Prefix() {
+ String version = "é".repeat(300);
+
+ byte[] field = VsrLoginCodec.sdkVersionField(version);
+
+ assertThat(field).hasSize(254);
+ assertThat(new String(field,
StandardCharsets.UTF_8)).isEqualTo("é".repeat(127));
+ }
+
+ @Test
+ void sdkVersionFieldKeepsSurrogatePairThatEndsExactlyAtU8Prefix() {
+ String version = "a".repeat(251) + "\uD83D\uDE00";
+ assertThat(version.getBytes(StandardCharsets.UTF_8)).hasSize(255);
+
+ byte[] field = VsrLoginCodec.sdkVersionField(version);
+
+ assertThat(field).hasSize(255);
+ assertThat(new String(field,
StandardCharsets.UTF_8)).isEqualTo(version);
+ }
+
+ @Test
+ void sdkVersionFieldDropsWholeSurrogatePairThatWouldStraddleU8Prefix() {
+ String version = "a".repeat(252) + "\uD83D\uDE00";
+
+ byte[] field = VsrLoginCodec.sdkVersionField(version);
+
+ assertThat(field).hasSize(252);
+ assertThat(new String(field,
StandardCharsets.UTF_8)).isEqualTo("a".repeat(252));
+ }
+
+ @Test
+ void rewriteUserLoginPrefixesCredentialsWithUtf8ByteLength() {
+ String username = "użytkownik";
+ String password = "hasło";
+ ByteBuf loginPayload = BytesSerializer.toBytes(username, "username");
+ loginPayload.writeBytes(BytesSerializer.toBytes(password, "password"));
+
+ ByteBuf body =
VsrLoginCodec.rewriteUserLogin(UnpooledByteBufAllocator.DEFAULT, loginPayload);
+ try {
+
assertThat(body.readIntLE()).isEqualTo(VsrLoginCodec.PROTOCOL_VERSION);
+ assertThat(readShortField(body)).isEqualTo(VsrLoginCodec.SDK_NAME);
+ assertThat(readShortField(body)).isNotEmpty();
+ assertThat(readShortField(body)).isEqualTo(username);
+ assertThat(readShortField(body)).isEqualTo(password);
+ assertThat(body.readIntLE()).isZero();
+ assertThat(body.isReadable()).isFalse();
+ } finally {
+ body.release();
+ loginPayload.release();
+ }
+ }
+
+ @Test
+ void rewriteUserLoginRejectsEmptyUsernameBeforeAllocating() {
+ ByteBuf loginPayload = Unpooled.buffer();
+ loginPayload.writeByte(0);
+ loginPayload.writeBytes(BytesSerializer.toBytes("secret", "password"));
+ CountingAllocator alloc = new CountingAllocator();
+
+ assertThatThrownBy(() -> VsrLoginCodec.rewriteUserLogin(alloc,
loginPayload))
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("username");
+ assertThat(alloc.allocations).isZero();
+ loginPayload.release();
+ }
+
+ @Test
+ void rewritePatLoginRejectsEmptyTokenBeforeAllocating() {
+ ByteBuf loginPayload = Unpooled.buffer();
+ loginPayload.writeByte(0);
+ CountingAllocator alloc = new CountingAllocator();
+
+ assertThatThrownBy(() -> VsrLoginCodec.rewritePatLogin(alloc,
loginPayload))
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("token");
+ assertThat(alloc.allocations).isZero();
+ loginPayload.release();
+ }
+
+ private static String readShortField(ByteBuf buffer) {
+ byte[] bytes = new byte[buffer.readUnsignedByte()];
+ buffer.readBytes(bytes);
+ return new String(bytes, StandardCharsets.UTF_8);
+ }
+
+ private static final class CountingAllocator extends
AbstractByteBufAllocator {
+ private int allocations;
+
+ @Override
+ protected ByteBuf newHeapBuffer(int initialCapacity, int maxCapacity) {
+ allocations++;
+ return Unpooled.buffer(initialCapacity, maxCapacity);
+ }
+
+ @Override
+ protected ByteBuf newDirectBuffer(int initialCapacity, int
maxCapacity) {
+ allocations++;
+ return Unpooled.directBuffer(initialCapacity, maxCapacity);
+ }
+
+ @Override
+ public boolean isDirectBufferPooled() {
+ return false;
+ }
+ }
+}
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..346b63bc3 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;
@@ -168,7 +171,7 @@ class BytesSerializerTest {
String input = "test";
// when
- ByteBuf result = BytesSerializer.toBytes(input);
+ ByteBuf result = BytesSerializer.toBytes(input, "name");
// then
assertThat(result.readByte()).isEqualTo((byte) 4); // length
@@ -178,16 +181,15 @@ class BytesSerializerTest {
}
@Test
- void shouldSerializeEmptyString() {
+ void shouldRejectEmptyString() {
// given
String input = "";
- // when
- ByteBuf result = BytesSerializer.toBytes(input);
-
- // then
- assertThat(result.readByte()).isEqualTo((byte) 0); // length = 0
- assertThat(result.readableBytes()).isEqualTo(0);
+ // when / then
+ assertThatThrownBy(() -> BytesSerializer.toBytes(input, "name"))
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("name")
+ .hasMessageContaining("0 bytes");
}
@Test
@@ -196,7 +198,7 @@ class BytesSerializerTest {
String input = "Hello世界";
// when
- ByteBuf result = BytesSerializer.toBytes(input);
+ ByteBuf result = BytesSerializer.toBytes(input, "name");
// then
byte[] expectedBytes = input.getBytes(StandardCharsets.UTF_8);
@@ -205,6 +207,55 @@ class BytesSerializerTest {
result.readBytes(stringBytes);
assertThat(stringBytes).isEqualTo(expectedBytes);
}
+
+ @Test
+ void shouldSerializeStringOfExactly255EncodedBytes() {
+ // given
+ String input = "世".repeat(85);
+
+ // when
+ ByteBuf result = BytesSerializer.toBytes(input, "name");
+
+ // then
+ assertThat(result.readUnsignedByte()).isEqualTo((short) 255);
+ assertThat(result.readableBytes()).isEqualTo(255);
+ }
+
+ @Test
+ void shouldRejectStringLongerThan255EncodedBytesEvenIfUnder255Chars() {
+ // given
+ String input = "あ".repeat(86);
+ assertThat(input.length()).isLessThan(255);
+
+ // when / then
+ assertThatThrownBy(() -> BytesSerializer.toBytes(input, "name"))
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("258");
+ }
+
+ @Test
+ void shouldNameTheRejectedField() {
+ assertThatThrownBy(() -> BytesSerializer.toBytes("", "username"))
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("Invalid username length");
+ }
+
+ @Test
+ void shouldApplyCallerBoundsToEncodedLength() {
+ // given: three chars, six bytes
+ String input = "ééé";
+
+ // when / then
+ assertThat(BytesSerializer.toBytes(input, "password", 3,
6).readUnsignedByte())
+ .isEqualTo((short) 6);
+ assertThatThrownBy(() -> BytesSerializer.toBytes(input,
"password", 3, 5))
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("password")
+ .hasMessageContaining("between 3 and 5");
+ assertThatThrownBy(() -> BytesSerializer.toBytes("ab", "password",
3, 5))
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("2 bytes");
+ }
}
@Nested
@@ -237,7 +288,45 @@ class BytesSerializerTest {
assertThat(result.readByte()).isEqualTo((byte) 11); // length =
"test-stream".length()
byte[] nameBytes = new byte[11];
result.readBytes(nameBytes);
- assertThat(new String(nameBytes)).isEqualTo("test-stream");
+ assertThat(new String(nameBytes,
StandardCharsets.UTF_8)).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 shouldMatchExpectedWireLayoutForNonAsciiNames(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("
", ""));
}
}
diff --git
a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/MessagesTcpClientTest.java
b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/MessagesTcpClientTest.java
index c644b50ac..6fc24de50 100644
---
a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/MessagesTcpClientTest.java
+++
b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/MessagesTcpClientTest.java
@@ -21,11 +21,19 @@ package org.apache.iggy.client.blocking.tcp;
import org.apache.iggy.client.blocking.IggyBaseClient;
import org.apache.iggy.client.blocking.MessagesClientBaseTest;
+import org.apache.iggy.consumergroup.Consumer;
+import org.apache.iggy.identifier.StreamId;
+import org.apache.iggy.identifier.TopicId;
import org.apache.iggy.message.Message;
import org.apache.iggy.message.Partitioning;
+import org.apache.iggy.message.PollingStrategy;
+import org.apache.iggy.topic.CompressionAlgorithm;
import org.junit.jupiter.api.Test;
+import java.math.BigInteger;
+import java.nio.charset.StandardCharsets;
import java.util.List;
+import java.util.Optional;
import static org.apache.iggy.TestConstants.STREAM_NAME;
import static org.apache.iggy.TestConstants.TOPIC_NAME;
@@ -59,4 +67,31 @@ class MessagesTcpClientTest extends MessagesClientBaseTest {
assertThat(secondResponse.confirmations().get(0).partitionId())
.isEqualTo(firstResponse.confirmations().get(0).partitionId());
}
+
+ /*
+ * TCP only: the HTTP client does not percent-encode path segments yet, so
+ * non-ASCII stream and topic names cannot be addressed over HTTP.
+ */
+ @Test
+ void shouldRoundTripNonAsciiNamesKeyAndPayload() {
+ // given
+ var streamId = StreamId.of("strumień-世界");
+ var topicId = TopicId.of("тема-日本語");
+ var stream = client.streams().createStream(streamId.getName());
+ trackStream(stream.id());
+ client.topics()
+ .createTopic(
+ streamId, 1L, CompressionAlgorithm.None,
BigInteger.ZERO, BigInteger.ZERO, topicId.getName());
+ String text = "wiadomość 世界 😀";
+
+ // when
+ messagesClient.sendMessages(streamId, topicId,
Partitioning.messagesKey("klucz-键"), List.of(Message.of(text)));
+ var polledMessages = messagesClient.pollMessages(
+ streamId, topicId, Optional.of(0L), Consumer.of(0L),
PollingStrategy.first(), 10L, false);
+
+ // then
+ assertThat(polledMessages.messages()).hasSize(1);
+ assertThat(new String(polledMessages.messages().get(0).payload(),
StandardCharsets.UTF_8))
+ .isEqualTo(text);
+ }
}
diff --git
a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/StreamTcpClientTest.java
b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/StreamTcpClientTest.java
index a3954f189..007efc4ec 100644
---
a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/StreamTcpClientTest.java
+++
b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/StreamTcpClientTest.java
@@ -21,6 +21,10 @@ package org.apache.iggy.client.blocking.tcp;
import org.apache.iggy.client.blocking.IggyBaseClient;
import org.apache.iggy.client.blocking.StreamClientBaseTest;
+import org.apache.iggy.identifier.StreamId;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
class StreamTcpClientTest extends StreamClientBaseTest {
@@ -28,4 +32,24 @@ class StreamTcpClientTest extends StreamClientBaseTest {
protected IggyBaseClient getClient() {
return TcpClientFactory.create(serverHost(), serverTcpPort());
}
+
+ /*
+ * TCP only: the HTTP client does not percent-encode path segments yet, so
+ * a non-ASCII name cannot be looked up over HTTP.
+ */
+ @Test
+ void shouldCreateAndFetchStreamWithNonAsciiName() {
+ // given
+ var name = "strumień-世界";
+
+ // when
+ var streamDetails = client.streams().createStream(name);
+ trackStream(streamDetails.id());
+ var streamByName = client.streams().getStream(StreamId.of(name));
+
+ // then
+ assertThat(streamDetails.name()).isEqualTo(name);
+ assertThat(streamByName).isPresent();
+ assertThat(streamByName.get().id()).isEqualTo(streamDetails.id());
+ }
}
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..6a73dcf91 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
@@ -19,10 +19,14 @@
package org.apache.iggy.identifier;
+import io.netty.buffer.ByteBuf;
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 +35,51 @@ public class IdentifierTest {
assertThatThrownBy(() -> new FakeIdentifier("foo",
123L)).isInstanceOf(IggyInvalidArgumentException.class);
}
+ @Test
+ void getSizeCountsEncodedBytesOfName() {
+ assertThat(new FakeIdentifier("世界", null).getSize()).isEqualTo(2 + 6);
+ }
+
+ @Test
+ void toBytesEncodesNameIdentifierAsKindLengthAndUtf8Bytes() {
+ byte[] expected = "世界".getBytes(StandardCharsets.UTF_8);
+
+ ByteBuf result = new FakeIdentifier("世界", null).toBytes();
+
+ assertThat(result.readableBytes()).isEqualTo(2 + expected.length);
+ assertThat(result.readByte()).isEqualTo((byte) 2);
+ assertThat(result.readUnsignedByte()).isEqualTo((short)
expected.length);
+ byte[] name = new byte[expected.length];
+ result.readBytes(name);
+ assertThat(name).isEqualTo(expected);
+ }
+
+ @Test
+ void toBytesEncodesNumericIdentifierAsKindLengthAndLittleEndianId() {
+ ByteBuf result = new FakeIdentifier(null, 7L).toBytes();
+
+ assertThat(result.readableBytes()).isEqualTo(6);
+ assertThat(result.readByte()).isEqualTo((byte) 1);
+ assertThat(result.readByte()).isEqualTo((byte) 4);
+ assertThat(result.readIntLE()).isEqualTo(7);
+ }
+
+ @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..260fd89d7 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,64 @@ 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
constructorThrowsIggyInvalidArgumentExceptionWhenValueExceeds255Bytes() {
+ assertThatThrownBy(() -> new
Partitioning(PartitioningKind.MessagesKey, new byte[256]))
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("256");
+ }
+
+ @Test
+ void
constructorThrowsIggyInvalidArgumentExceptionWhenMessagesKeyValueIsEmpty() {
+ assertThatThrownBy(() -> new
Partitioning(PartitioningKind.MessagesKey, new byte[0]))
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("1..255");
+ }
+
+ @Test
+ void
constructorThrowsIggyInvalidArgumentExceptionWhenBalancedValueIsNotEmpty() {
+ assertThatThrownBy(() -> new Partitioning(PartitioningKind.Balanced,
new byte[] {1}))
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("Balanced");
+ }
+
+ @ParameterizedTest
+ @ValueSource(ints = {0, 3, 5})
+ void
constructorThrowsIggyInvalidArgumentExceptionWhenPartitionIdValueIsNotFourBytes(int
length) {
+ assertThatThrownBy(() -> new
Partitioning(PartitioningKind.PartitionId, new byte[length]))
+ .isInstanceOf(IggyInvalidArgumentException.class)
+ .hasMessageContaining("must be 4 bytes");
+ }
+
+ @Test
+ void constructorThrowsIggyInvalidArgumentExceptionWhenKindOrValueIsNull() {
+ assertThatThrownBy(() -> new Partitioning(null, new
byte[0])).isInstanceOf(IggyInvalidArgumentException.class);
+ assertThatThrownBy(() -> new
Partitioning(PartitioningKind.MessagesKey, null))
+ .isInstanceOf(IggyInvalidArgumentException.class);
+ }
+
+ @Test
+ void partitionIdThrowsIggyInvalidArgumentExceptionWhenIdIsNull() {
+ assertThatThrownBy(() ->
Partitioning.partitionId(null)).isInstanceOf(IggyInvalidArgumentException.class);
+ }
+
+ @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";
