Add connect data function with headers
@@ -22,6 +22,7 @@ import io.apicurio.registry.serde.AbstractKafkaDeserializer;
import io.apicurio.registry.serde.AbstractKafkaSerializer;
import io.apicurio.registry.utils.IoUtil;
+import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.connect.data.Schema;
@@ -133,6 +134,15 @@ public class SerdeBasedConverter<S, T> implements Converter, Closeable {
return toSchemaAndValue(result);
}
+ @Override
+ public SchemaAndValue toConnectData(String topic, Headers headers, byte[] bytes) {
+ T result = deserializer.deserialize(topic, headers, bytes);
+ if (result == null) {
+ return SchemaAndValue.NULL;
+ }
+ return toSchemaAndValue(result);
+ }
+
public void setSerializer(Serializer<T> serializer) {
this.serializer = Objects.requireNonNull(serializer);
}
