BE: Closes #71 Messages: Show headers duplicates

api/src/main/java/io/kafbat/ui/emitter/MessageFilters.java

@@ -123,7 +123,7 @@ private static CelCompiler createCompiler() {
         "timestampMs", SimpleType.INT,
         "keyAsText", SimpleType.STRING,
         "valueAsText", SimpleType.STRING,
-        "headers", MapType.create(SimpleType.STRING, SimpleType.STRING),
+        "headers", MapType.create(SimpleType.STRING, SimpleType.DYN),
         "key", SimpleType.DYN,
         "value", SimpleType.DYN
     );

api/src/main/java/io/kafbat/ui/serdes/ConsumerRecordDeserializer.java

@@ -6,8 +6,10 @@
 import java.time.Instant;
 import java.time.OffsetDateTime;
 import java.time.ZoneId;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.function.UnaryOperator;
 import lombok.RequiredArgsConstructor;
@@ -63,14 +65,16 @@ private static TimestampTypeEnum mapToTimestampType(TimestampType timestampType)
   }
 
   private void fillHeaders(TopicMessageDTO message, ConsumerRecord<Bytes, Bytes> rec) {
-    Map<String, String> headers = new HashMap<>();
+    Map<String, List<String>> headersMap = new HashMap<>();
     rec.headers().iterator()
-        .forEachRemaining(header ->
-            headers.put(
-                header.key(),
-                header.value() != null ? new String(header.value()) : null
-            ));
-    message.setHeaders(headers);
+        .forEachRemaining(header -> {
+          String key = header.key();
+          String value = header.value() != null ? new String(header.value()) : null;
+          headersMap.computeIfAbsent(key, k -> new ArrayList<>()).add(value);
+        });
+    Map<String, Object> finalHeadersMap = new HashMap<>();
+    headersMap.forEach((key, values) -> finalHeadersMap.put(key, values.size() == 1 ? values.get(0) : values));
+    message.setHeaders(finalHeadersMap);
   }
 
   private void fillKey(TopicMessageDTO message, ConsumerRecord<Bytes, Bytes> rec) {

api/src/main/java/io/kafbat/ui/serdes/ProducerRecordCreator.java

@@ -1,6 +1,8 @@
 package io.kafbat.ui.serdes;
 
+import io.kafbat.ui.exception.ValidationException;
 import io.kafbat.ui.serde.api.Serde;
+import java.util.List;
 import java.util.Map;
 import javax.annotation.Nullable;
 import lombok.RequiredArgsConstructor;
@@ -19,7 +21,7 @@ public ProducerRecord<byte[], byte[]> create(String topic,
                                                @Nullable Integer partition,
                                                @Nullable String key,
                                                @Nullable String value,
-                                               @Nullable Map<String, String> headers) {
+                                               @Nullable Map<String, Object> headers) {
     return new ProducerRecord<>(
         topic,
         partition,
@@ -29,10 +31,23 @@ public ProducerRecord<byte[], byte[]> create(String topic,
     );
   }
 
-  private Iterable<Header> createHeaders(Map<String, String> clientHeaders) {
+  private Iterable<Header> createHeaders(Map<String, Object> clientHeaders) {
     RecordHeaders headers = new RecordHeaders();
-    clientHeaders.forEach((k, v) -> headers.add(new RecordHeader(k, v.getBytes())));
+    clientHeaders.forEach((k, v) -> {
+      if (v instanceof List<?> valueList) {
+        valueList.forEach(value -> headers.add(new RecordHeader(k, valueToBytes(value))));
+      } else {
+        headers.add(new RecordHeader(k, valueToBytes(v)));
+      }
+    });
     return headers;
   }
 
+  private byte[] valueToBytes(Object value) {
+    if (value instanceof List<?> || value instanceof Map<?, ?>) {
+      throw new ValidationException("Header values can only be string or list of strings");
+    }
+    return value != null ? String.valueOf(value).getBytes() : null;
+  }
+
 }

api/src/test/java/io/kafbat/ui/serdes/ConsumerRecordDeserializerTest.java

@@ -1,13 +1,17 @@
 package io.kafbat.ui.serdes;
 
 import static io.kafbat.ui.serde.api.DeserializeResult.Type.STRING;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
 
 import io.kafbat.ui.model.TopicMessageDTO;
 import io.kafbat.ui.serde.api.DeserializeResult;
 import io.kafbat.ui.serde.api.Serde;
+import java.util.List;
 import java.util.Map;
 import java.util.function.UnaryOperator;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -22,9 +26,51 @@ void dataMaskingAppliedOnDeserializedMessage() {
     Serde.Deserializer deser = (headers, data) -> new DeserializeResult("test", STRING, Map.of());
 
     var recordDeser = new ConsumerRecordDeserializer("test", deser, "test", deser, "test", deser, deser, maskerMock);
-    recordDeser.deserialize(new ConsumerRecord<>("t", 1, 1L, Bytes.wrap("t".getBytes()), Bytes.wrap("t".getBytes())));
+    recordDeser.deserialize(record());
 
     verify(maskerMock).apply(any(TopicMessageDTO.class));
   }
 
+  @Test
+  void deserializeWithMultipleHeaderValues() {
+    UnaryOperator<TopicMessageDTO> maskerMock = mock();
+    when(maskerMock.apply(any(TopicMessageDTO.class))).thenAnswer(invocation -> invocation.getArgument(0));
+    Serde.Deserializer deser = (headers, data) -> new DeserializeResult("test", STRING, Map.of());
+
+    var recordDeser = new ConsumerRecordDeserializer("test", deser, "test", deser, "test", deser, deser, maskerMock);
+    ConsumerRecord<Bytes, Bytes> record = record();
+    record.headers().add("headerKey", "headerValue1".getBytes());
+    record.headers().add("headerKey", "headerValue2".getBytes());
+    TopicMessageDTO message = recordDeser.deserialize(record);
+
+    Map<String, Object> headers = message.getHeaders();
+    assertEquals(1, headers.size());
+    assertTrue(headers.get("headerKey") instanceof List);
+    assertEquals(List.of("headerValue1", "headerValue2"), headers.get("headerKey"));
+  }
+
+  @Test
+  void deserializeWithMixedSingleAndMultipleHeaderValues() {
+    UnaryOperator<TopicMessageDTO> maskerMock = mock();
+    when(maskerMock.apply(any(TopicMessageDTO.class))).thenAnswer(invocation -> invocation.getArgument(0));
+    Serde.Deserializer deser = (headers, data) -> new DeserializeResult("test", STRING, Map.of());
+
+    var recordDeser = new ConsumerRecordDeserializer("test", deser, "test", deser, "test", deser, deser, maskerMock);
+    ConsumerRecord<Bytes, Bytes> record = record();
+    record.headers().add("headerKey1", "singleValue".getBytes());
+    record.headers().add("headerKey2", "multiValue1".getBytes());
+    record.headers().add("headerKey2", "multiValue2".getBytes());
+    TopicMessageDTO message = recordDeser.deserialize(record);
+
+    Map<String, Object> headers = message.getHeaders();
+    assertEquals("singleValue", headers.get("headerKey1"));
+    assertTrue(headers.get("headerKey2") instanceof List);
+    assertEquals(2, ((List<?>) headers.get("headerKey2")).size());
+    assertEquals(List.of("multiValue1", "multiValue2"), headers.get("headerKey2"));
+  }
+
+  private ConsumerRecord<Bytes, Bytes> record() {
+    return new ConsumerRecord<>("t", 1, 1L, Bytes.wrap("t".getBytes()), Bytes.wrap("t".getBytes()));
+  }
+
 }

api/src/test/java/io/kafbat/ui/serdes/ProducerRecordCreatorTest.java

@@ -0,0 +1,51 @@
+package io.kafbat.ui.serdes;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.Mockito.mock;
+
+import io.kafbat.ui.exception.ValidationException;
+import io.kafbat.ui.serde.api.Serde;
+import java.util.List;
+import java.util.Map;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.header.internals.RecordHeader;
+import org.junit.jupiter.api.Test;
+
+class ProducerRecordCreatorTest {
+
+  @Test
+  void createWithHeaders() {
+    Serde.Serializer keySerializer = mock(Serde.Serializer.class);
+    Serde.Serializer valueSerializer = mock(Serde.Serializer.class);
+
+    ProducerRecordCreator recordCreator = new ProducerRecordCreator(keySerializer, valueSerializer);
+    Map<String, Object> headersMap = Map.of(
+        "headerKey1", "headerValue1",
+        "headerKey2", List.of("headerValue2", "headerValue3")
+    );
+    ProducerRecord<byte[], byte[]> record = recordCreator.create("topic", 1, "key", "value", headersMap);
+
+    assertNotNull(record.headers());
+    assertEquals(3, record.headers().toArray().length);
+    assertThat(record.headers()).containsExactlyInAnyOrder(
+        new RecordHeader("headerKey1", "headerValue1".getBytes()),
+        new RecordHeader("headerKey2", "headerValue2".getBytes()),
+        new RecordHeader("headerKey2", "headerValue3".getBytes())
+    );
+  }
+
+  @Test
+  void createWithInvalidHeaderValue() {
+    Serde.Serializer keySerializer = mock(Serde.Serializer.class);
+    Serde.Serializer valueSerializer = mock(Serde.Serializer.class);
+
+    ProducerRecordCreator recordCreator = new ProducerRecordCreator(keySerializer, valueSerializer);
+    Map<String, Object> invalidHeaders = Map.of("headerKey", Map.of("invalid", "value"));
+
+    assertThrows(ValidationException.class, () ->
+        recordCreator.create("topic", 1, "key", "value", invalidHeaders));
+  }
+}

contract/src/main/resources/swagger/kafbat-ui-api.yaml

@@ -2919,8 +2919,9 @@ components:
           type: string
         headers:
           type: object
+          description: header values can be string or list of strings (for keys with multiple values)
           additionalProperties:
-            type: string
+            type: object
         partition:
           type: integer
         offset:
@@ -2948,8 +2949,9 @@ components:
           nullable: true
         headers:
           type: object
+          description: header values can be string or list of strings (for keys with multiple values)
           additionalProperties:
-            type: string
+            type: object
         content:
           type: string
           nullable: true
@@ -3038,8 +3040,9 @@ components:
           type: string
         headers:
           type: object
+          description: header values can be string or list of strings (for keys with multiple values)
           additionalProperties:
-            type: string
+            type: object
         content:
           type: string
         keyFormat:

frontend/src/components/Topics/Topic/Messages/Filters/InfoModal.tsx

@@ -54,6 +54,14 @@ const InfoModal: React.FC<InfoModalProps> = ({ toggleIsOpen }) => {
             record.headers[&apos;k2&apos;] == &apos;v2&apos;
           </code>
         </S.ListItem>
+        <S.ListItem>
+          <code>
+            &quot;v1&quot; in (type(record.headers.k1) == list ?
+            record.headers.k1 : [record.headers.k1])
+          </code>{' '}
+          - in case a header can hold either a single value (string) or multiple
+          values (list of strings)
+        </S.ListItem>
       </ol>
       <Flexbox justifyContent="center" margin="20px 0 0 0">
         <Button

frontend/src/components/Topics/Topic/Messages/MessageContent/MessageContent.tsx

@@ -11,7 +11,7 @@ type Tab = 'key' | 'content' | 'headers';
 export interface MessageContentProps {
   messageKey?: string;
   messageContent?: string;
-  headers?: { [key: string]: string | undefined };
+  headers?: { [key: string]: string | string[] | undefined };
   timestamp?: Date;
   timestampType?: TopicMessageTimestampTypeEnum;
   keySize?: number;

frontend/src/components/Topics/Topic/Messages/MessageContent/__tests__/MessageContent.spec.tsx

@@ -17,7 +17,7 @@ const setupWrapper = (props?: Partial<MessageContentProps>) => {
         <MessageContent
           messageKey='"test-key"'
           messageContent='{"data": "test"}'
-          headers={{ header: 'test' }}
+          headers={{ header1: 'test', header2: ['value1', 'value2'] }}
           timestamp={new Date(0)}
           timestampType={TopicMessageTimestampTypeEnum.CREATE_TIME}
           keySerde="SchemaRegistry"