beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "peay (JIRA)" <>
Subject [jira] [Commented] (BEAM-1573) KafkaIO does not allow using Kafka serializers and deserializers
Date Thu, 02 Mar 2017 18:38:45 GMT


peay commented on BEAM-1573:

My concern is for both source and sink.

I'd like to be able to use custom {{org.apache.kafka.common.serialization.Serializer,Deserializer}}s.
An example is
for working with setups where Kafka topics contain Avro serialized using an Avro schema registry.
This uses a {{Serializer/Deserializer<Object>}} but I also have similar Kafka serializers
with arbitrary types.

The interfaces of the encoding/decoding methods in {{org.apache.kafka.common.serialization.Serializer,Deserializer}}
- {{serialize(String topic, byte[] data)}}
- {{deserialize(String topic, byte[] data)}}.

I would like to be able to support a syntax like this:



.withCustomKafkaValueDeserializerAndCoder(new SomeCustomKafkaDeserializer(), AvroCoder.of(xxx))

.withCustomKafkaKeyDeserializerAndCoder(new SomeCustomKafkaDeserializer()), AvroCoder.of(xxx))





.withCustomKafkaValueSerializer(new SomeCustomDeserializer())

.withCustomKafkaKeySerializer(new SomeCustomDeserializer()))

In both case, Kafka would use the custom serializer/deserializer directly.

Now, why is it hard to express currently? KafkaIO seems to be implemented differently for
read and write, so let us consider the two cases. I have a working patch for the above syntax,
that is straightforward for writes, but requires a bunch of changes for reads...

For write, the Coder is wrapped into an actual {{org.apache.kafka.common.serialization.Serializer}}
through {{CoderBasedKafkaSerializer}}. I can make a {{CustomCoder}}, but still have to pass
it manually the topic name. Also, we end up with a wrapper for a Kafka serializer, wrapped
in a Coder, itself wrapped in a Kafka serializer. 

Reads are implemented differently. I am not sure why? Instead of wrapping the coders into
a Kafka deserializer, everything is hard wired to use `byte[]` Kafka consumer. Then, KakfaIO
calls the coder after data has been returned by the consumer. Here also, one can make a {{CustomCoder}}.
This won't work if a list of topics is used as input to KafkaIO, and still requires to pass
in the topic name manually when there's only here. In the example snippet above, I also include
a second argument that is a coder, to use with {{setCoder}} for setting up the rest of the

In both cases, wrapping the Kafka serializer into the Coder is also not obvious because Kafka
serializers have a {{configure}} method to give them access to the consumer/producer config,
so this possibly needs to be emulated in the coder wrapper.

What do you think? 

> KafkaIO does not allow using Kafka serializers and deserializers
> ----------------------------------------------------------------
>                 Key: BEAM-1573
>                 URL:
>             Project: Beam
>          Issue Type: Bug
>          Components: sdk-java-extensions
>    Affects Versions: 0.4.0, 0.5.0
>            Reporter: peay
>            Assignee: Raghu Angadi
>            Priority: Minor
> KafkaIO does not allow to override the serializer and deserializer settings of the Kafka
consumer and producers it uses internally. Instead, it allows to set a `Coder`, and has a
simple Kafka serializer/deserializer wrapper class that calls the coder.
> I appreciate that allowing to use Beam coders is good and consistent with the rest of
the system. However, is there a reason to completely disallow to use custom Kafka serializers
> This is a limitation when working with an Avro schema registry for instance, which requires
custom serializers. One can write a `Coder` that wraps a custom Kafka serializer, but that
means two levels of un-necessary wrapping.
> In addition, the `Coder` abstraction is not equivalent to Kafka's `Serializer` which
gets the topic name as input. Using a `Coder` wrapper would require duplicating the output
topic setting in the argument to `KafkaIO` and when building the wrapper, which is not elegant
and error prone.

This message was sent by Atlassian JIRA

View raw message