You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by pi...@apache.org on 2026/01/30 12:56:03 UTC
(iggy) branch non_string_header_key updated (df3348fb4 -> 8c6f4c9ad)
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a change to branch non_string_header_key
in repository https://gitbox.apache.org/repos/asf/iggy.git
from df3348fb4 merge
new 7e10a9633 Add boundary checks for node, java, cli, make key/value private
add 393b0f6fa feat(connectors): extend published stats (#2640)
add 9f0057685 chore(repo): bump Rust version to 1.93.0 (#2644)
new 8c6f4c9ad Merge branch 'master' into non_string_header_key
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/actions/utils/docker-buildx/action.yml | 4 +-
Dockerfile | 2 +-
bdd/java/Dockerfile | 2 +-
bdd/python/Dockerfile | 2 +-
bdd/rust/Dockerfile | 2 +-
core/ai/mcp/Dockerfile | 2 +-
core/bench/dashboard/server/Dockerfile | 2 +-
.../src/cli/binary_message/poll_messages.rs | 4 +-
core/cli/src/args/message.rs | 24 ++-
core/common/src/lib.rs | 35 ++--
core/common/src/types/message/user_headers.rs | 49 ++---
core/common/src/utils/mod.rs | 35 ++--
.../{server/src => common/src/utils}/versioning.rs | 27 +--
core/connectors/runtime/Dockerfile | 2 +-
core/connectors/runtime/src/stats.rs | 30 ++-
.../tests/cli/message/test_message_poll_command.rs | 2 +-
.../tests/cli/message/test_message_send_command.rs | 7 +-
core/integration/tests/state/mod.rs | 5 +-
core/server/Dockerfile | 2 +-
core/server/src/bootstrap.rs | 2 +-
core/server/src/lib.rs | 8 +-
core/server/src/main.rs | 2 +-
core/server/src/shard/builder.rs | 5 +-
core/server/src/shard/mod.rs | 7 +-
core/server/src/shard/system/info.rs | 5 +-
core/server/src/state/file.rs | 5 +-
.../message-headers/typed-headers/consumer/main.rs | 4 +-
.../java/org/apache/iggy/message/HeaderValue.java | 203 ++++++++++++++++++++-
.../main/java/org/apache/iggy/message/Message.java | 2 +-
.../org/apache/iggy/serde/BytesDeserializer.java | 8 +-
.../org/apache/iggy/serde/BytesSerializer.java | 4 +-
.../client/blocking/tcp/BytesSerializerTest.java | 9 +-
.../apache/iggy/serde/BytesDeserializerTest.java | 4 +-
foreign/node/src/wire/message/header.utils.ts | 11 ++
rust-toolchain.toml | 2 +-
35 files changed, 384 insertions(+), 135 deletions(-)
rename core/{server/src => common/src/utils}/versioning.rs (94%)
(iggy) 01/02: Add boundary checks for node, java, cli, make key/value private
Posted by pi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch non_string_header_key
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 7e10a96332b3190321681417358b7e2718a77e84
Author: spetz <pi...@gmail.com>
AuthorDate: Fri Jan 30 13:53:10 2026 +0100
Add boundary checks for node, java, cli, make key/value private
---
.../src/cli/binary_message/poll_messages.rs | 4 +-
core/cli/src/args/message.rs | 24 ++-
core/common/src/types/message/user_headers.rs | 49 ++---
.../tests/cli/message/test_message_poll_command.rs | 2 +-
.../tests/cli/message/test_message_send_command.rs | 7 +-
.../message-headers/typed-headers/consumer/main.rs | 4 +-
.../java/org/apache/iggy/message/HeaderValue.java | 203 ++++++++++++++++++++-
.../main/java/org/apache/iggy/message/Message.java | 2 +-
.../org/apache/iggy/serde/BytesDeserializer.java | 8 +-
.../org/apache/iggy/serde/BytesSerializer.java | 4 +-
.../client/blocking/tcp/BytesSerializerTest.java | 9 +-
.../apache/iggy/serde/BytesDeserializerTest.java | 4 +-
foreign/node/src/wire/message/header.utils.ts | 11 ++
13 files changed, 284 insertions(+), 47 deletions(-)
diff --git a/core/binary_protocol/src/cli/binary_message/poll_messages.rs b/core/binary_protocol/src/cli/binary_message/poll_messages.rs
index dd210ddb8..7b6725512 100644
--- a/core/binary_protocol/src/cli/binary_message/poll_messages.rs
+++ b/core/binary_protocol/src/cli/binary_message/poll_messages.rs
@@ -88,7 +88,7 @@ impl PollMessagesCmd {
match HashMap::<HeaderKey, HeaderValue>::from_bytes(user_headers.clone()) {
Ok(headers) => headers
.iter()
- .map(|(k, v)| (k.clone(), v.kind))
+ .map(|(k, v)| (k.clone(), v.kind()))
.collect::<Vec<_>>(),
Err(e) => {
tracing::error!("Failed to parse user headers, error: {e}");
@@ -146,7 +146,7 @@ impl PollMessagesCmd {
.as_ref()
.map(|h| {
h.get(key)
- .filter(|v| v.kind == *kind)
+ .filter(|v| v.kind() == *kind)
.map(|v| v.to_string_value())
.unwrap_or_default()
})
diff --git a/core/cli/src/args/message.rs b/core/cli/src/args/message.rs
index 5157722fe..9b70df702 100644
--- a/core/cli/src/args/message.rs
+++ b/core/cli/src/args/message.rs
@@ -120,15 +120,14 @@ pub(crate) struct SendMessagesArgs {
/// Parse Header Key, Kind and Value from the string separated by a ':'
fn parse_key_val(s: &str) -> Result<(HeaderKey, HeaderValue), IggyError> {
- let lower = s.to_lowercase();
- let parts = lower.split(':').collect::<Vec<_>>();
+ let parts = s.splitn(3, ':').collect::<Vec<_>>();
if parts.len() != 3 {
return Err(IggyError::InvalidFormat);
}
let key = HeaderKey::from_str(parts[0])?;
- let kind = HeaderKind::from_str(parts[1])?;
+ let kind = HeaderKind::from_str(&parts[1].to_lowercase())?;
let value_str = parts[2];
let value = match kind {
@@ -351,4 +350,23 @@ mod tests {
let result = parse_key_val("key:uint8:69.42");
assert!(result.is_err());
}
+
+ #[test]
+ fn parse_key_val_should_preserve_value_case() {
+ let expected_value = "HelloWorld";
+ let result = parse_key_val(&format!("key:string:{expected_value}"));
+ assert!(result.is_ok());
+ let (_, value) = result.unwrap();
+ assert_eq!(value.as_str().unwrap(), expected_value);
+ }
+
+ #[test]
+ fn parse_key_val_should_preserve_colons_in_value() {
+ let expected_value = "http://example.com:8080";
+ let result = parse_key_val(&format!("url:string:{expected_value}"));
+ assert!(result.is_ok());
+ let (key, value) = result.unwrap();
+ assert_eq!(key, HeaderKey::from_str("url").unwrap());
+ assert_eq!(value.as_str().unwrap(), expected_value);
+ }
}
diff --git a/core/common/src/types/message/user_headers.rs b/core/common/src/types/message/user_headers.rs
index 353bbc072..aaec4ffe0 100644
--- a/core/common/src/types/message/user_headers.rs
+++ b/core/common/src/types/message/user_headers.rs
@@ -146,15 +146,30 @@ pub struct ValueMarker;
#[serde_as]
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
pub struct HeaderField<T> {
- /// The type of value stored in this header field.
- pub kind: HeaderKind,
- /// The raw bytes of the value, encoded according to `kind`.
+ kind: HeaderKind,
#[serde_as(as = "Base64")]
- pub value: Bytes,
+ value: Bytes,
#[serde(skip)]
_marker: PhantomData<T>,
}
+impl<T> HeaderField<T> {
+ /// Returns the kind of this header field.
+ pub fn kind(&self) -> HeaderKind {
+ self.kind
+ }
+
+ /// Returns a clone of the raw bytes value.
+ pub fn value(&self) -> Bytes {
+ self.value.clone()
+ }
+
+ /// Returns a reference to the raw bytes value.
+ pub fn as_bytes(&self) -> &[u8] {
+ &self.value
+ }
+}
+
/// Indicates the type of value stored in a [`HeaderField`].
///
/// This enum is used to track what type of data is stored in the header's raw bytes,
@@ -923,14 +938,14 @@ impl BytesSerializable for HashMap<HeaderKey, HeaderValue> {
let mut bytes = BytesMut::new();
for (key, value) in self {
- bytes.put_u8(key.kind.as_code());
+ bytes.put_u8(key.kind().as_code());
#[allow(clippy::cast_possible_truncation)]
- bytes.put_u32_le(key.value.len() as u32);
- bytes.put_slice(&key.value);
- bytes.put_u8(value.kind.as_code());
+ bytes.put_u32_le(key.as_bytes().len() as u32);
+ bytes.put_slice(key.as_bytes());
+ bytes.put_u8(value.kind().as_code());
#[allow(clippy::cast_possible_truncation)]
- bytes.put_u32_le(value.value.len() as u32);
- bytes.put_slice(&value.value);
+ bytes.put_u32_le(value.as_bytes().len() as u32);
+ bytes.put_slice(value.as_bytes());
}
bytes.freeze()
@@ -1005,16 +1020,8 @@ impl BytesSerializable for HashMap<HeaderKey, HeaderValue> {
position += value_length;
headers.insert(
- HeaderKey {
- kind: key_kind,
- value: Bytes::from(key_value),
- _marker: PhantomData,
- },
- HeaderValue {
- kind: value_kind,
- value: Bytes::from(value_value),
- _marker: PhantomData,
- },
+ HeaderKey::new_unchecked(key_kind, &key_value),
+ HeaderValue::new_unchecked(value_kind, &value_value),
);
}
@@ -1026,7 +1033,7 @@ pub fn get_user_headers_size(headers: &Option<HashMap<HeaderKey, HeaderValue>>)
let mut size = 0;
if let Some(headers) = headers {
for (key, value) in headers {
- size += 1 + 4 + key.value.len() as u32 + 1 + 4 + value.value.len() as u32;
+ size += 1 + 4 + key.as_bytes().len() as u32 + 1 + 4 + value.as_bytes().len() as u32;
}
}
Some(size)
diff --git a/core/integration/tests/cli/message/test_message_poll_command.rs b/core/integration/tests/cli/message/test_message_poll_command.rs
index 867a8aca1..7d7532912 100644
--- a/core/integration/tests/cli/message/test_message_poll_command.rs
+++ b/core/integration/tests/cli/message/test_message_poll_command.rs
@@ -199,7 +199,7 @@ impl IggyCmdTestCase for TestMessagePollCmd {
"Header: {}",
self.headers.0.to_string_value()
)))
- .stdout(contains(self.headers.1.kind.to_string()))
+ .stdout(contains(self.headers.1.kind().to_string()))
.stdout(contains(self.headers.1.to_string_value()).count(self.message_count))
}
diff --git a/core/integration/tests/cli/message/test_message_send_command.rs b/core/integration/tests/cli/message/test_message_send_command.rs
index fea3498e5..97ed4fe66 100644
--- a/core/integration/tests/cli/message/test_message_send_command.rs
+++ b/core/integration/tests/cli/message/test_message_send_command.rs
@@ -110,7 +110,12 @@ impl TestMessageSendCmd {
header
.iter()
.map(|(k, v)| {
- format!("{}:{}:{}", k.to_string_value(), v.kind, v.to_string_value())
+ format!(
+ "{}:{}:{}",
+ k.to_string_value(),
+ v.kind(),
+ v.to_string_value()
+ )
})
.collect::<Vec<_>>()
.join(","),
diff --git a/examples/rust/src/message-headers/typed-headers/consumer/main.rs b/examples/rust/src/message-headers/typed-headers/consumer/main.rs
index ce9a3be76..496f9b09d 100644
--- a/examples/rust/src/message-headers/typed-headers/consumer/main.rs
+++ b/examples/rust/src/message-headers/typed-headers/consumer/main.rs
@@ -59,9 +59,9 @@ fn handle_message(message: &IggyMessage) -> Result<(), Box<dyn Error>> {
for (key, value) in &headers_map {
info!(
" key: [kind={}, value={}] -> value: [kind={}, value={}]",
- key.kind,
+ key.kind(),
key.to_string_value(),
- value.kind,
+ value.kind(),
value.to_string_value()
);
}
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/HeaderValue.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/HeaderValue.java
index 90b3bf053..dba7918ef 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/HeaderValue.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/message/HeaderValue.java
@@ -22,15 +22,212 @@ package org.apache.iggy.message;
import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
import java.util.Base64;
-public record HeaderValue(HeaderKind kind, String value) {
+public record HeaderValue(HeaderKind kind, byte[] value) {
@JsonCreator
public static HeaderValue fromJson(
@JsonProperty("kind") HeaderKind kind, @JsonProperty("value") String base64Value) {
byte[] decodedValue = Base64.getDecoder().decode(base64Value);
- String stringValue = new String(decodedValue, StandardCharsets.UTF_8);
- return new HeaderValue(kind, stringValue);
+ return new HeaderValue(kind, decodedValue);
+ }
+
+ public static HeaderValue fromString(String val) {
+ if (val.isEmpty() || val.length() > 255) {
+ throw new IllegalArgumentException("Value has incorrect size, must be between 1 and 255");
+ }
+ return new HeaderValue(HeaderKind.String, val.getBytes(StandardCharsets.UTF_8));
+ }
+
+ public static HeaderValue fromBool(boolean val) {
+ return new HeaderValue(HeaderKind.Bool, new byte[] {(byte) (val ? 1 : 0)});
+ }
+
+ public static HeaderValue fromInt8(byte val) {
+ return new HeaderValue(HeaderKind.Int8, new byte[] {val});
+ }
+
+ public static HeaderValue fromInt16(short val) {
+ ByteBuffer buffer = ByteBuffer.allocate(2).order(ByteOrder.LITTLE_ENDIAN);
+ buffer.putShort(val);
+ return new HeaderValue(HeaderKind.Int16, buffer.array());
+ }
+
+ public static HeaderValue fromInt32(int val) {
+ ByteBuffer buffer = ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN);
+ buffer.putInt(val);
+ return new HeaderValue(HeaderKind.Int32, buffer.array());
+ }
+
+ public static HeaderValue fromInt64(long val) {
+ ByteBuffer buffer = ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN);
+ buffer.putLong(val);
+ return new HeaderValue(HeaderKind.Int64, buffer.array());
+ }
+
+ public static HeaderValue fromUint8(short val) {
+ if (val < 0 || val > 255) {
+ throw new IllegalArgumentException("Value must be between 0 and 255");
+ }
+ return new HeaderValue(HeaderKind.Uint8, new byte[] {(byte) val});
+ }
+
+ public static HeaderValue fromUint16(int val) {
+ if (val < 0 || val > 65535) {
+ throw new IllegalArgumentException("Value must be between 0 and 65535");
+ }
+ ByteBuffer buffer = ByteBuffer.allocate(2).order(ByteOrder.LITTLE_ENDIAN);
+ buffer.putShort((short) val);
+ return new HeaderValue(HeaderKind.Uint16, buffer.array());
+ }
+
+ public static HeaderValue fromUint32(long val) {
+ if (val < 0 || val > 4294967295L) {
+ throw new IllegalArgumentException("Value must be between 0 and 4294967295");
+ }
+ ByteBuffer buffer = ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN);
+ buffer.putInt((int) val);
+ return new HeaderValue(HeaderKind.Uint32, buffer.array());
+ }
+
+ public static HeaderValue fromFloat32(float val) {
+ ByteBuffer buffer = ByteBuffer.allocate(4).order(ByteOrder.LITTLE_ENDIAN);
+ buffer.putFloat(val);
+ return new HeaderValue(HeaderKind.Float32, buffer.array());
+ }
+
+ public static HeaderValue fromFloat64(double val) {
+ ByteBuffer buffer = ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN);
+ buffer.putDouble(val);
+ return new HeaderValue(HeaderKind.Float64, buffer.array());
+ }
+
+ public static HeaderValue fromRaw(byte[] val) {
+ if (val.length == 0 || val.length > 255) {
+ throw new IllegalArgumentException("Value has incorrect size, must be between 1 and 255");
+ }
+ return new HeaderValue(HeaderKind.Raw, val);
+ }
+
+ public String asString() {
+ if (kind != HeaderKind.String) {
+ throw new IllegalStateException("Header value is not a string, kind: " + kind);
+ }
+ return new String(value, StandardCharsets.UTF_8);
+ }
+
+ public boolean asBool() {
+ if (kind != HeaderKind.Bool) {
+ throw new IllegalStateException("Header value is not a bool, kind: " + kind);
+ }
+ return value[0] == 1;
+ }
+
+ public byte asInt8() {
+ if (kind != HeaderKind.Int8) {
+ throw new IllegalStateException("Header value is not an int8, kind: " + kind);
+ }
+ return value[0];
+ }
+
+ public short asInt16() {
+ if (kind != HeaderKind.Int16) {
+ throw new IllegalStateException("Header value is not an int16, kind: " + kind);
+ }
+ return ByteBuffer.wrap(value).order(ByteOrder.LITTLE_ENDIAN).getShort();
+ }
+
+ public int asInt32() {
+ if (kind != HeaderKind.Int32) {
+ throw new IllegalStateException("Header value is not an int32, kind: " + kind);
+ }
+ return ByteBuffer.wrap(value).order(ByteOrder.LITTLE_ENDIAN).getInt();
+ }
+
+ public long asInt64() {
+ if (kind != HeaderKind.Int64) {
+ throw new IllegalStateException("Header value is not an int64, kind: " + kind);
+ }
+ return ByteBuffer.wrap(value).order(ByteOrder.LITTLE_ENDIAN).getLong();
+ }
+
+ public short asUint8() {
+ if (kind != HeaderKind.Uint8) {
+ throw new IllegalStateException("Header value is not a uint8, kind: " + kind);
+ }
+ return (short) (value[0] & 0xFF);
+ }
+
+ public int asUint16() {
+ if (kind != HeaderKind.Uint16) {
+ throw new IllegalStateException("Header value is not a uint16, kind: " + kind);
+ }
+ return (ByteBuffer.wrap(value).order(ByteOrder.LITTLE_ENDIAN).getShort() & 0xFFFF);
+ }
+
+ public long asUint32() {
+ if (kind != HeaderKind.Uint32) {
+ throw new IllegalStateException("Header value is not a uint32, kind: " + kind);
+ }
+ return (ByteBuffer.wrap(value).order(ByteOrder.LITTLE_ENDIAN).getInt() & 0xFFFFFFFFL);
+ }
+
+ public float asFloat32() {
+ if (kind != HeaderKind.Float32) {
+ throw new IllegalStateException("Header value is not a float32, kind: " + kind);
+ }
+ return ByteBuffer.wrap(value).order(ByteOrder.LITTLE_ENDIAN).getFloat();
+ }
+
+ public double asFloat64() {
+ if (kind != HeaderKind.Float64) {
+ throw new IllegalStateException("Header value is not a float64, kind: " + kind);
+ }
+ return ByteBuffer.wrap(value).order(ByteOrder.LITTLE_ENDIAN).getDouble();
+ }
+
+ public byte[] asRaw() {
+ return value;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ HeaderValue that = (HeaderValue) o;
+ return kind == that.kind && Arrays.equals(value, that.value);
+ }
+
+ @Override
+ public int hashCode() {
+ int result = kind.hashCode();
+ result = 31 * result + Arrays.hashCode(value);
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return switch (kind) {
+ case String -> asString();
+ case Bool -> String.valueOf(asBool());
+ case Int8 -> String.valueOf(asInt8());
+ case Int16 -> String.valueOf(asInt16());
+ case Int32 -> String.valueOf(asInt32());
+ case Int64 -> String.valueOf(asInt64());
+ case Uint8 -> String.valueOf(asUint8());
+ case Uint16 -> String.valueOf(asUint16());
+ case Uint32 -> String.valueOf(asUint32());
+ case Float32 -> String.valueOf(asFloat32());
+ case Float64 -> String.valueOf(asFloat64());
+ case Raw, Int128, Uint64, Uint128 -> Base64.getEncoder().encodeToString(value);
+ };
}
}
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 8e4d4bc21..144103d10 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
@@ -113,7 +113,7 @@ public record Message(
long size = 0L;
for (Map.Entry<HeaderKey, HeaderValue> entry : userHeaders.entrySet()) {
byte[] keyBytes = entry.getKey().value();
- byte[] valueBytes = entry.getValue().value().getBytes();
+ byte[] valueBytes = entry.getValue().value();
size += 1L + 4L + keyBytes.length + 1L + 4L + valueBytes.length;
}
return size;
diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesDeserializer.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesDeserializer.java
index ef94e1590..fcd16ca55 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesDeserializer.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/BytesDeserializer.java
@@ -210,11 +210,11 @@ public final class BytesDeserializer {
var userHeaderValueKindCode = userHeadersBuffer.readUnsignedByte();
var userHeaderValueLength = userHeadersBuffer.readUnsignedIntLE();
- String userHeaderValue = userHeadersBuffer
- .readCharSequence(toInt(userHeaderValueLength), StandardCharsets.UTF_8)
- .toString();
+ byte[] userHeaderValueBytes = new byte[toInt(userHeaderValueLength)];
+ userHeadersBuffer.readBytes(userHeaderValueBytes);
headers.put(
- userHeaderKey, new HeaderValue(HeaderKind.fromCode(userHeaderValueKindCode), userHeaderValue));
+ userHeaderKey,
+ new HeaderValue(HeaderKind.fromCode(userHeaderValueKindCode), userHeaderValueBytes));
}
userHeaders = headers;
}
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 cc6f6bd11..20301ecdb 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
@@ -134,8 +134,8 @@ public final class BytesSerializer {
HeaderValue value = entry.getValue();
buffer.writeByte(value.kind().asCode());
- buffer.writeIntLE(value.value().length());
- buffer.writeBytes(value.value().getBytes());
+ buffer.writeIntLE(value.value().length);
+ buffer.writeBytes(value.value());
}
return buffer;
}
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 99514dc4d..3453f1863 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
@@ -27,7 +27,6 @@ import org.apache.iggy.identifier.ConsumerId;
import org.apache.iggy.identifier.StreamId;
import org.apache.iggy.message.BytesMessageId;
import org.apache.iggy.message.HeaderKey;
-import org.apache.iggy.message.HeaderKind;
import org.apache.iggy.message.HeaderValue;
import org.apache.iggy.message.Message;
import org.apache.iggy.message.MessageHeader;
@@ -459,7 +458,7 @@ class BytesSerializerTest {
void shouldSerializeSingleHeader() {
// given
Map<HeaderKey, HeaderValue> headers = new HashMap<>();
- headers.put(HeaderKey.fromString("key1"), new HeaderValue(HeaderKind.Raw, "value1"));
+ headers.put(HeaderKey.fromString("key1"), HeaderValue.fromRaw("value1".getBytes()));
// when
ByteBuf result = BytesSerializer.toBytes(headers);
@@ -481,8 +480,8 @@ class BytesSerializerTest {
void shouldSerializeMultipleHeaders() {
// given
Map<HeaderKey, HeaderValue> headers = new HashMap<>();
- headers.put(HeaderKey.fromString("k1"), new HeaderValue(HeaderKind.Raw, "v1"));
- headers.put(HeaderKey.fromString("k2"), new HeaderValue(HeaderKind.String, "v2"));
+ headers.put(HeaderKey.fromString("k1"), HeaderValue.fromRaw("v1".getBytes()));
+ headers.put(HeaderKey.fromString("k2"), HeaderValue.fromString("v2"));
// when
ByteBuf result = BytesSerializer.toBytes(headers);
@@ -523,7 +522,7 @@ class BytesSerializerTest {
// given
var messageId = new BytesMessageId(new byte[16]);
Map<HeaderKey, HeaderValue> userHeaders = new HashMap<>();
- userHeaders.put(HeaderKey.fromString("key"), new HeaderValue(HeaderKind.Raw, "val"));
+ userHeaders.put(HeaderKey.fromString("key"), HeaderValue.fromRaw("val".getBytes()));
// Calculate user headers size
ByteBuf headersBuf = BytesSerializer.toBytes(userHeaders);
diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/serde/BytesDeserializerTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/serde/BytesDeserializerTest.java
index 001cfcdbe..36a6d5ea3 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/serde/BytesDeserializerTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/serde/BytesDeserializerTest.java
@@ -382,8 +382,8 @@ class BytesDeserializerTest {
// then
assertThat(message.userHeaders()).hasSize(1);
- assertThat(message.userHeaders().get(HeaderKey.fromString("key")).value())
- .isEqualTo("val");
+ assertThat(message.userHeaders().get(HeaderKey.fromString("key")).asRaw())
+ .isEqualTo("val".getBytes());
}
@Test
diff --git a/foreign/node/src/wire/message/header.utils.ts b/foreign/node/src/wire/message/header.utils.ts
index 2b2be17a5..34b1486f7 100644
--- a/foreign/node/src/wire/message/header.utils.ts
+++ b/foreign/node/src/wire/message/header.utils.ts
@@ -278,6 +278,7 @@ export const deserializeHeaderValue = (
* @param p - Buffer containing serialized headers
* @param pos - Starting position in the buffer
* @returns Object with bytes read and deserialized header data
+ * @throws Error if header key or value length is invalid (must be 1-255)
*/
export const deserializeHeader = (
p: Buffer,
@@ -285,11 +286,21 @@ export const deserializeHeader = (
): ParsedHeaderDeserialized => {
const _keyKind = p.readUInt8(pos);
const keyLength = p.readUInt32LE(pos + 1);
+ if (keyLength < 1 || keyLength > 255) {
+ throw new Error(
+ `Invalid header key length: ${keyLength}, must be between 1 and 255`,
+ );
+ }
const key = p.subarray(pos + 5, pos + 5 + keyLength).toString();
pos += 5 + keyLength;
const valueKind = p.readUInt8(pos);
const valueLength = p.readUInt32LE(pos + 1);
+ if (valueLength < 1 || valueLength > 255) {
+ throw new Error(
+ `Invalid header value length: ${valueLength}, must be between 1 and 255`,
+ );
+ }
const value = deserializeHeaderValue(
valueKind,
p.subarray(pos + 5, pos + 5 + valueLength),
(iggy) 02/02: Merge branch 'master' into non_string_header_key
Posted by pi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
piotr pushed a commit to branch non_string_header_key
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 8c6f4c9ad77d973b1b05e84bc1e20aa00cabda63
Merge: 7e10a9633 9f0057685
Author: spetz <pi...@gmail.com>
AuthorDate: Fri Jan 30 13:55:42 2026 +0100
Merge branch 'master' into non_string_header_key
.github/actions/utils/docker-buildx/action.yml | 4 +--
Dockerfile | 2 +-
bdd/java/Dockerfile | 2 +-
bdd/python/Dockerfile | 2 +-
bdd/rust/Dockerfile | 2 +-
core/ai/mcp/Dockerfile | 2 +-
core/bench/dashboard/server/Dockerfile | 2 +-
core/common/src/lib.rs | 35 ++++++++++++----------
core/common/src/utils/mod.rs | 35 ++++++++++++----------
.../{server/src => common/src/utils}/versioning.rs | 27 +++--------------
core/connectors/runtime/Dockerfile | 2 +-
core/connectors/runtime/src/stats.rs | 30 +++++++++++++++----
core/integration/tests/state/mod.rs | 5 ++--
core/server/Dockerfile | 2 +-
core/server/src/bootstrap.rs | 2 +-
core/server/src/lib.rs | 8 ++---
core/server/src/main.rs | 2 +-
core/server/src/shard/builder.rs | 5 ++--
core/server/src/shard/mod.rs | 7 +++--
core/server/src/shard/system/info.rs | 5 ++--
core/server/src/state/file.rs | 5 ++--
rust-toolchain.toml | 2 +-
22 files changed, 100 insertions(+), 88 deletions(-)