hubcio commented on code in PR #4068:
URL: https://github.com/apache/iggy/pull/4068#discussion_r3942120930
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/UsersTcpClient.java:
##########
@@ -155,20 +157,23 @@ public CompletableFuture<IdentityInfo> login(String
username, String password) {
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);
Review Comment:
warning: these bytes never reach the wire - `VsrLoginCodec.rewriteUserLogin`
reads username and password only, then hardcodes `writeIntLE(0)` for the
context. drop version and context here, or thread `context` through the codec.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java:
##########
@@ -210,9 +214,12 @@ public static ByteBuf toBytes(TopicPermissions
permissions) {
}
public static ByteBuf toBytes(String value) {
- int bufferLength = 1 + value.length();
- ByteBuf buffer = Unpooled.buffer(bufferLength);
byte[] stringBytes = value.getBytes(StandardCharsets.UTF_8);
+ if (stringBytes.length > MAX_U8_STRING_LENGTH) {
+ throw new IggyInvalidArgumentException("String must be at most " +
MAX_U8_STRING_LENGTH
+ + " bytes when UTF-8 encoded, got " + stringBytes.length);
+ }
+ ByteBuf buffer = Unpooled.buffer(1 + stringBytes.length);
buffer.writeByte(stringBytes.length);
Review Comment:
nit: no lower bound - `toBytes("")` still writes `[0x00]`, and the server
rejects a zero-length wire name, so `createStream("")` builds a frame it always
refuses. pre-existing, and adding a floor means flipping
`shouldSerializeEmptyString`.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/StreamsTcpClient.java:
##########
@@ -102,12 +99,12 @@ public CompletableFuture<List<StreamBase>> getStreams() {
@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());
+ var nameBytes = BytesSerializer.toBytes(name);
+ var payload = Unpooled.buffer(idBytes.readableBytes() +
nameBytes.readableBytes());
payload.writeBytes(idBytes);
- payload.writeBytes(BytesSerializer.toBytes(name));
+ payload.writeBytes(nameBytes);
// 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()));
Review Comment:
simplification: `toBytes` of an empty map returns `EMPTY_BUFFER`, so this
writes nothing. the comment above already marks the extension point - same at
`UsersTcpClient.java:125`.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/identifier/Identifier.java:
##########
@@ -23,10 +23,15 @@
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;
Review Comment:
simplification: nothing outside this file uses `MAX_NAME_LENGTH`, and the
same holds for `Partitioning.MAX_MESSAGES_KEY_LENGTH`. both are new, so making
them private breaks nothing.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java:
##########
@@ -63,6 +63,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;
Review Comment:
warning: 255 is the wire cap, but the server bounds username at 3..50,
password at 3..100 and PAT name at 3..30. a 200-byte username passes here and
then dies at the server - bound them at the call sites.
##########
foreign/java/java-sdk/src/test/java/org/apache/iggy/message/MessageTest.java:
##########
@@ -72,6 +73,14 @@ void
ofCreatesMessageWhenGivenMessageHeaderPayloadAndNullUserHeaders() {
assertThat(message.userHeaders().size()).isEqualTo(0);
}
+ @Test
+ void ofEncodesStringPayloadAsUtf8() {
Review Comment:
warning: this passes against pre-PR code on any UTF-8-default JVM, and the
Test task sets no `defaultCharacterEncoding`, so CI never runs the case it's
named for. same at `PartitioningTest.java:81`. fork the test JVM with
`file.encoding=ISO-8859-1`.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java:
##########
@@ -23,8 +23,20 @@
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 Partitioning {
+ if (value.length > MAX_MESSAGES_KEY_LENGTH) {
Review Comment:
warning: this only checks the ceiling, and it checks it for every kind. an
empty `MessagesKey` and a `PartitionId` that isn't 4 bytes both pass here and
get rejected by the server - switch on `kind` instead.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java:
##########
@@ -38,10 +50,15 @@ public static Partitioning partitionId(Long id) {
}
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);
Review Comment:
warning: this changes message-key routing - `MessagesTcpClient` hashes these
bytes client-side, so a non-ASCII key lands on a different partition after the
upgrade on any JVM whose default charset isn't UTF-8. correct change, but it
needs a release note.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesSerializer.java:
##########
@@ -210,9 +214,12 @@ public static ByteBuf toBytes(TopicPermissions
permissions) {
}
public static ByteBuf toBytes(String value) {
- int bufferLength = 1 + value.length();
- ByteBuf buffer = Unpooled.buffer(bufferLength);
byte[] stringBytes = value.getBytes(StandardCharsets.UTF_8);
+ if (stringBytes.length > MAX_U8_STRING_LENGTH) {
Review Comment:
nit: seven fields funnel through here - stream, topic and group names, PAT
name, username, password, PAT token - so this message tells the caller nothing
about which one broke. `checkFieldLength` below takes a field label for exactly
that reason.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/identifier/Identifier.java:
##########
@@ -68,13 +80,21 @@ public String getName() {
return name;
}
+ /**
+ * The name as UTF-8 wire bytes, encoded once at construction; {@code
null} for a numeric
+ * identifier. The array is shared rather than copied, so callers must not
modify it.
+ */
+ @Nullable public byte[] getEncodedName() {
Review Comment:
nit: new public API on a public abstract class that hands back the live
array. a caller that mutates it keeps the length prefix valid and silently
sends a different name than `getName()` reports.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java:
##########
@@ -94,15 +103,22 @@ static long readSessionEpoch(ByteBuf registerBody) {
private static void writeVersionInfo(ByteBuf body) {
body.writeIntLE(PROTOCOL_VERSION);
writeShortField(body, SDK_NAME.getBytes(StandardCharsets.UTF_8));
Review Comment:
nit: `SDK_NAME` and the version are both constants, re-encoded on every
login and reconnect. hoist them to `private static final byte[]`, keeping
`sdkVersionField` package-visible for the test.
##########
foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/BytesSerializerTest.java:
##########
@@ -239,6 +267,44 @@ void shouldSerializeStringIdentifier() {
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) {
Review Comment:
nit: the name says server wire format, but the body compares hand-written
hex with nothing tying it to the Rust encoder. rename to
`shouldMatchExpectedWireLayoutForNonAsciiNames`, or move the parity claim into
`bdd/`.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java:
##########
@@ -94,15 +103,22 @@ static long readSessionEpoch(ByteBuf registerBody) {
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,
sdkVersionField(IggyVersion.getInstance().getVersion()));
}
- private static String sdkVersion() {
- String version = IggyVersion.getInstance().getVersion();
- if (version == null || version.isEmpty()) {
- return "unknown";
- }
- return version.length() > 255 ? version.substring(0, 255) : version;
+ /**
+ * 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;
+ ByteBuffer encoded = ByteBuffer.allocate(MAX_SHORT_FIELD_LENGTH);
+ StandardCharsets.UTF_8
+ .newEncoder()
+ .onMalformedInput(CodingErrorAction.REPLACE)
+ .onUnmappableCharacter(CodingErrorAction.REPLACE)
+ .encode(CharBuffer.wrap(value), encoded, true);
Review Comment:
nit: the encoder contract wants `flush(out)` after `encode(in, out, true)`.
it works today only because the UTF-8 encoder's `implFlush` is a no-op - append
`.flush(encoded)`.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java:
##########
@@ -38,10 +50,15 @@ public static Partitioning partitionId(Long id) {
}
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) {
Review Comment:
simplification: the compact constructor applies this same bound four lines
down. drop the check here and let the constructor throw -
`hasMessageContaining("300")` still passes.
##########
foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/BytesSerializerTest.java:
##########
@@ -239,6 +267,44 @@ void shouldSerializeStringIdentifier() {
result.readBytes(nameBytes);
assertThat(new String(nameBytes)).isEqualTo("test-stream");
Review Comment:
nit: `new String(nameBytes)` decodes on the platform default, the same
conflation this PR is removing. pass `StandardCharsets.UTF_8`.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/ConsumerGroupsTcpClient.java:
##########
@@ -106,11 +106,13 @@ 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());
+ var nameBytes = BytesSerializer.toBytes(name);
+ var payload = Unpooled.buffer(
Review Comment:
simplification: `toBytes(streamId)` followed by two `writeBytes` calls does
the same thing without the manual sizing, matching `:61` and `:84` in this
file. same shape at `StreamsTcpClient.java:104`.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/vsr/VsrLoginCodec.java:
##########
@@ -94,15 +103,22 @@ static long readSessionEpoch(ByteBuf registerBody) {
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,
sdkVersionField(IggyVersion.getInstance().getVersion()));
}
- private static String sdkVersion() {
- String version = IggyVersion.getInstance().getVersion();
- if (version == null || version.isEmpty()) {
- return "unknown";
- }
- return version.length() > 255 ? version.substring(0, 255) : version;
+ /**
+ * 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) {
Review Comment:
simplification: `getBytes(UTF_8)` already replaces unpaired surrogates, so
this reduces to encode once, return if it's 255 bytes or under, else walk back
off the continuation bytes. drops the flush issue too.
##########
foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java:
##########
@@ -23,8 +23,20 @@
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 Partitioning {
Review Comment:
nit: `new Partitioning(kind, null)` throws a bare NPE at the length check
instead of `IggyInvalidArgumentException`. same shape at `partitionId(null)`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
