Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Serde: Enable header modifications by custom serializer #509

Open
wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Serde: Enable header modifications by custom serializer #509

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
fe9dc76
Initial proposition to enable header modifications by custom serializer
MDzaja Jul 17, 2024
2c55546
add comments, change variable names, optimize config
MDzaja Jul 28, 2024
b177272
Merge branch 'main' into feature/issue#346
MDzaja Jul 29, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
17 changes: 11 additions & 6 deletions api/src/main/java/io/kafbat/ui/serdes/ProducerRecordCreator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand All @@ -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;
}

Expand Down
20 changes: 16 additions & 4 deletions api/src/main/java/io/kafbat/ui/serdes/SerdeInstance.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why we'd need a proxy here?

Copy link
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We need a proxy here because the Serializer interface now has multiple methods, so we cannot use a simple lambda for wrapping (as was done previously). A proxy provides a generic, scalable, and clean solution to wrap all methods of the Serializer interface dynamically. This ensures that every method invocation on the interface is consistently wrapped with the necessary logic (e.g., wrapWithClassloader).

While it’s possible to use an explicit or anonymous class implementation to achieve the same goal, using a proxy reduces boilerplate and makes the code easier to maintain in the long run. For comparison, here’s how an anonymous wrapper class could look:

public Serde.Serializer serializer(String topic, Serde.Target type) {
    return wrapWithClassloader(() -> {
        var serializer = serde.serializer(topic, type);
        return new Serde.Serializer() {
            @Override
            public byte[] serialize(String input) {
                return wrapWithClassloader(() -> serializer.serialize(input));
            }

            @Override
            public byte[] serialize(String input, Headers headers) {
                return wrapWithClassloader(() -> serializer.serialize(input, headers));
            }
        };
    });
}

While this approach works, it introduces repetitive boilerplate for every method, which becomes harder to manage as the interface evolves.

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) {
Expand Down
14 changes: 14 additions & 0 deletions serde-api/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>

Expand All @@ -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>
Expand Down
5 changes: 5 additions & 0 deletions serde-api/src/main/java/io/kafbat/ui/serde/api/Serde.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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);
}
}

/**
Expand Down
Loading