diff --git a/api/src/main/java/io/kafbat/ui/serdes/ProducerRecordCreator.java b/api/src/main/java/io/kafbat/ui/serdes/ProducerRecordCreator.java index 359c871d6..6f4db7a77 100644 --- a/api/src/main/java/io/kafbat/ui/serdes/ProducerRecordCreator.java +++ b/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 create(String topic, @Nullable String key, @Nullable String value, @Nullable Map 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
createHeaders(Map clientHeaders) { + private Headers createHeaders(Map 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; } diff --git a/api/src/main/java/io/kafbat/ui/serdes/SerdeInstance.java b/api/src/main/java/io/kafbat/ui/serdes/SerdeInstance.java index 7c1826257..791535081 100644 --- a/api/src/main/java/io/kafbat/ui/serdes/SerdeInstance.java +++ b/api/src/main/java/io/kafbat/ui/serdes/SerdeInstance.java @@ -3,6 +3,7 @@ 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; @@ -10,6 +11,7 @@ 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) { diff --git a/serde-api/pom.xml b/serde-api/pom.xml index 3c2e3b3a1..7ae589f08 100644 --- a/serde-api/pom.xml +++ b/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"> + + kafbat-ui + io.kafbat.ui + 0.0.1-SNAPSHOT + + 4.0.0 jar @@ -12,6 +18,14 @@ 17 + + + org.apache.kafka + kafka-clients + ${kafka-clients.version} + + + ossrh diff --git a/serde-api/src/main/java/io/kafbat/ui/serde/api/Serde.java b/serde-api/src/main/java/io/kafbat/ui/serde/api/Serde.java index 74705b1b6..45a46814b 100644 --- a/serde-api/src/main/java/io/kafbat/ui/serde/api/Serde.java +++ b/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.
Note: this input is not formatted in any way. */ byte[] serialize(String input); + + default byte[] serialize(String input, Headers headers) { + return serialize(input); + } } /**