Fix tests

Author: spetz

foreign/csharp/Iggy_SDK/Contracts/MessageResponse.cs

@@ -40,6 +40,7 @@ public sealed class MessageResponse
     /// <summary>
     ///     Headers defined by the user.
     /// </summary>
+    [JsonPropertyName("user_headers")]
     [JsonConverter(typeof(UserHeadersConverter))]
     public Dictionary<HeaderKey, HeaderValue>? UserHeaders { get; init; }
 }

foreign/csharp/Iggy_SDK/Contracts/Tcp/TcpContracts.cs

@@ -554,12 +554,12 @@ private static byte[] GetHeadersBytes(Dictionary<HeaderKey, HeaderValue>? header
             return [];
         }
 
-        var headersLength = headers.Sum(header => 1 + 4 + header.Key.Value.Length + 1 + 4 + header.Value.Value.Length);
+        var headersLength = headers.Sum(kvp => 1 + 4 + kvp.Key.Value.Length + 1 + 4 + kvp.Value.Value.Length);
         Span<byte> headersBytes = stackalloc byte[headersLength];
         var position = 0;
-        foreach (var (headerKey, headerValue) in headers)
+        foreach (var kvp in headers)
         {
-            var headerBytes = GetBytesFromHeader(headerKey, headerValue);
+            var headerBytes = GetBytesFromHeader(kvp.Key, kvp.Value);
             headerBytes.CopyTo(headersBytes[position..(position + headerBytes.Length)]);
             position += headerBytes.Length;
         }

foreign/csharp/Iggy_SDK/Headers/HeaderEntry.cs

@@ -0,0 +1,25 @@
+// 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.
+
+namespace Apache.Iggy.Headers;
+
+/// <summary>
+///     Represents a single header entry with key and value.
+/// </summary>
+/// <param name="Key">The header key.</param>
+/// <param name="Value">The header value.</param>
+public readonly record struct HeaderEntry(HeaderKey Key, HeaderValue Value);

foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs

@@ -55,7 +55,11 @@ internal HttpMessageStream(HttpClient httpClient)
         _jsonSerializerOptions = new JsonSerializerOptions
         {
             PropertyNamingPolicy = JsonNamingPolicy.SnakeCaseLower,
-            Converters = { new JsonStringEnumConverter(JsonNamingPolicy.SnakeCaseLower) }
+            Converters =
+            {
+                new JsonStringEnumConverter(JsonNamingPolicy.SnakeCaseLower),
+                new JsonConverters.UserHeadersConverter()
+            }
         };
     }
 

foreign/csharp/Iggy_SDK/JsonConverters/UserHeadersConverter.cs

@@ -32,7 +32,7 @@ internal class UserHeadersConverter : JsonConverter<Dictionary<HeaderKey, Header
 
         if (reader.TokenType != JsonTokenType.StartArray)
         {
-            throw new JsonException("Expected start of array for headers.");
+            throw new JsonException($"Expected start of array for headers but got {reader.TokenType}.");
         }
 
         var result = new Dictionary<HeaderKey, HeaderValue>();
@@ -41,7 +41,7 @@ internal class UserHeadersConverter : JsonConverter<Dictionary<HeaderKey, Header
         {
             if (reader.TokenType == JsonTokenType.EndArray)
             {
-                break;
+                return result.Count == 0 ? null : result;
             }
 
             if (reader.TokenType != JsonTokenType.StartObject)
@@ -70,7 +70,7 @@ internal class UserHeadersConverter : JsonConverter<Dictionary<HeaderKey, Header
                 switch (propertyName)
                 {
                     case "key":
-                        key = ReadHeaderField(ref reader);
+                        key = ReadHeaderKey(ref reader);
                         break;
                     case "value":
                         value = ReadHeaderValue(ref reader);
@@ -89,7 +89,7 @@ internal class UserHeadersConverter : JsonConverter<Dictionary<HeaderKey, Header
         return result;
     }
 
-    private static HeaderKey ReadHeaderField(ref Utf8JsonReader reader)
+    private static HeaderKey ReadHeaderKey(ref Utf8JsonReader reader)
     {
         if (reader.TokenType != JsonTokenType.StartObject)
         {
@@ -212,18 +212,18 @@ public override void Write(Utf8JsonWriter writer, Dictionary<HeaderKey, HeaderVa
 
         writer.WriteStartArray();
 
-        foreach (var (headerKey, headerValue) in value)
+        foreach (var kvp in value)
         {
             writer.WriteStartObject();
 
             writer.WriteStartObject("key");
-            writer.WriteString("kind", GetHeaderKindString(headerKey.Kind));
-            writer.WriteBase64String("value", headerKey.Value);
+            writer.WriteString("kind", GetHeaderKindString(kvp.Key.Kind));
+            writer.WriteBase64String("value", kvp.Key.Value);
             writer.WriteEndObject();
 
             writer.WriteStartObject("value");
-            writer.WriteString("kind", GetHeaderKindString(headerValue.Kind));
-            writer.WriteBase64String("value", headerValue.Value);
+            writer.WriteString("kind", GetHeaderKindString(kvp.Value.Kind));
+            writer.WriteBase64String("value", kvp.Value.Value);
             writer.WriteEndObject();
 
             writer.WriteEndObject();

foreign/csharp/Iggy_SDK/Mappers/BinaryMapper.cs

@@ -454,11 +454,8 @@ private static Dictionary<HeaderKey, HeaderValue> MapHeaders(ReadOnlySpan<byte>
             ReadOnlySpan<byte> value = payload[position..(position + valueLength)];
             position += valueLength;
 
-            headers.Add(new HeaderKey { Kind = keyKind, Value = keyValue }, new HeaderValue
-            {
-                Kind = valueKind,
-                Value = value.ToArray()
-            });
+            headers[new HeaderKey { Kind = keyKind, Value = keyValue }] =
+                new HeaderValue { Kind = valueKind, Value = value.ToArray() };
         }
 
         return headers;

foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Message.java

@@ -23,6 +23,9 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonSetter;
 import com.fasterxml.jackson.annotation.Nulls;
+import org.apache.iggy.serde.Base64Serializer;
+import org.apache.iggy.serde.UserHeadersSerializer;
+import tools.jackson.databind.annotation.JsonSerialize;
 
 import java.math.BigInteger;
 import java.util.Base64;
@@ -31,7 +34,10 @@
 import java.util.List;
 import java.util.Map;
 
-public record Message(MessageHeader header, byte[] payload, Map<HeaderKey, HeaderValue> userHeaders) {
+public record Message(
+        MessageHeader header,
+        @JsonSerialize(using = Base64Serializer.class) byte[] payload,
+        @JsonSerialize(using = UserHeadersSerializer.class) Map<HeaderKey, HeaderValue> userHeaders) {
     @JsonCreator
     public static Message fromJson(
             @JsonProperty("header") MessageHeader header,

foreign/java/java-sdk/src/main/java/org/apache/iggy/message/Partitioning.java

@@ -20,11 +20,14 @@
 package org.apache.iggy.message;
 
 import org.apache.commons.lang3.ArrayUtils;
+import org.apache.iggy.serde.Base64Serializer;
+import tools.jackson.databind.annotation.JsonSerialize;
 
 import java.nio.ByteBuffer;
 
-public record Partitioning(PartitioningKind kind, byte[] value) {
-
+public record Partitioning(
+        PartitioningKind kind,
+        @JsonSerialize(using = Base64Serializer.class) byte[] value) {
     public static Partitioning balanced() {
         return new Partitioning(PartitioningKind.Balanced, new byte[] {});
     }

foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/Base64Serializer.java

@@ -0,0 +1,35 @@
+/*
+ * 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.serde;
+
+import tools.jackson.core.JacksonException;
+import tools.jackson.core.JsonGenerator;
+import tools.jackson.databind.SerializationContext;
+import tools.jackson.databind.ValueSerializer;
+
+import java.util.Base64;
+
+public class Base64Serializer extends ValueSerializer<byte[]> {
+
+    @Override
+    public void serialize(byte[] value, JsonGenerator gen, SerializationContext ctxt) throws JacksonException {
+        gen.writeString(Base64.getEncoder().encodeToString(value));
+    }
+}

foreign/java/java-sdk/src/main/java/org/apache/iggy/serde/UserHeadersSerializer.java

@@ -0,0 +1,44 @@
+/*
+ * 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.serde;
+
+import org.apache.iggy.message.HeaderEntry;
+import org.apache.iggy.message.HeaderKey;
+import org.apache.iggy.message.HeaderValue;
+import tools.jackson.core.JacksonException;
+import tools.jackson.core.JsonGenerator;
+import tools.jackson.databind.SerializationContext;
+import tools.jackson.databind.ValueSerializer;
+
+import java.util.Map;
+
+public class UserHeadersSerializer extends ValueSerializer<Map<HeaderKey, HeaderValue>> {
+
+    @Override
+    public void serialize(Map<HeaderKey, HeaderValue> headers, JsonGenerator gen, SerializationContext ctxt)
+            throws JacksonException {
+        gen.writeStartArray();
+        for (Map.Entry<HeaderKey, HeaderValue> entry : headers.entrySet()) {
+            ctxt.findValueSerializer(HeaderEntry.class)
+                    .serialize(new HeaderEntry(entry.getKey(), entry.getValue()), gen, ctxt);
+        }
+        gen.writeEndArray();
+    }
+}