Serde: Enable header serialization

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

@@ -5,7 +5,7 @@
 import javax.annotation.Nullable;
 import lombok.RequiredArgsConstructor;
 import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.header.Headers;
 import org.apache.kafka.common.header.internals.RecordHeader;
 import org.apache.kafka.common.header.internals.RecordHeaders;
 
@@ -20,18 +20,23 @@ public ProducerRecord<byte[], byte[]> create(String topic,
                                                @Nullable String key,
                                                @Nullable String value,
                                                @Nullable Map<String, String> headers) {
+
+    Headers kafkaHeaders = createHeaders(headers);
+
     return new ProducerRecord<>(
         topic,
         partition,
-        key == null ? null : keySerializer.serialize(key),
-        value == null ? null : valuesSerializer.serialize(value),
-        headers == null ? null : createHeaders(headers)
+        key == null ? null : keySerializer.serialize(key, kafkaHeaders),
+        value == null ? null : valuesSerializer.serialize(value, kafkaHeaders),
+        kafkaHeaders
     );
   }
 
-  private Iterable<Header> createHeaders(Map<String, String> clientHeaders) {
+  private Headers createHeaders(Map<String, String> clientHeaders) {
     RecordHeaders headers = new RecordHeaders();
-    clientHeaders.forEach((k, v) -> headers.add(new RecordHeader(k, v.getBytes())));
+    if (clientHeaders != null) {
+      clientHeaders.forEach((k, v) -> headers.add(new RecordHeader(k, v.getBytes())));
+    }
     return headers;
   }
 

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

@@ -3,13 +3,15 @@
 import io.kafbat.ui.serde.api.SchemaDescription;
 import io.kafbat.ui.serde.api.Serde;
 import java.io.Closeable;
+import java.lang.reflect.InvocationTargetException;
 import java.util.Optional;
 import java.util.function.Supplier;
 import java.util.regex.Pattern;
 import javax.annotation.Nullable;
 import lombok.Getter;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.cglib.proxy.Proxy;
 
 @Slf4j
 @RequiredArgsConstructor
@@ -78,10 +80,20 @@ public boolean canDeserialize(String topic, Serde.Target type) {
   }
 
   public Serde.Serializer serializer(String topic, Serde.Target type) {
-    return wrapWithClassloader(() -> {
-      var serializer = serde.serializer(topic, type);
-      return input -> wrapWithClassloader(() -> serializer.serialize(input));
-    });
+    var serializer = serde.serializer(topic, type);
+    // Create a dynamic proxy instance for the Serde.Serializer interface
+    return (Serde.Serializer) Proxy.newProxyInstance(
+        classLoader,
+        new Class<?>[] { Serde.Serializer.class },
+        (proxy, method, args) -> wrapWithClassloader(() -> { // Invocation handler to wrap method calls
+          try {
+            // Invoke the actual serializer method with the provided arguments
+            return method.invoke(serializer, args);
+          } catch (IllegalAccessException | InvocationTargetException e) {
+            throw new RuntimeException("Error invoking serializer method", e.getCause());
+          }
+        })
+    );
   }
 
   public Serde.Deserializer deserializer(String topic, Serde.Target type) {

serde-api/pom.xml

@@ -4,6 +4,12 @@
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
 
+    <parent>
+        <artifactId>kafbat-ui</artifactId>
+        <groupId>io.kafbat.ui</groupId>
+        <version>0.0.1-SNAPSHOT</version>
+    </parent>
+
     <modelVersion>4.0.0</modelVersion>
     <packaging>jar</packaging>
 
@@ -12,6 +18,14 @@
         <maven.compiler.target>17</maven.compiler.target>
     </properties>
 
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+            <version>${kafka-clients.version}</version>
+        </dependency>
+    </dependencies>
+
     <distributionManagement>
         <snapshotRepository>
             <id>ossrh</id>

serde-api/src/main/java/io/kafbat/ui/serde/api/Serde.java

@@ -2,6 +2,7 @@
 
 import java.io.Closeable;
 import java.util.Optional;
+import org.apache.kafka.common.header.Headers;
 
 /**
  * Main interface of  serialization/deserialization logic.
@@ -96,6 +97,10 @@ interface Serializer {
      * @param input string entered by user into UI text field.<br/> Note: this input is not formatted in any way.
      */
     byte[] serialize(String input);
+
+    default byte[] serialize(String input, Headers headers) {
+      return serialize(input);
+    }
   }
 
   /**