Automatic DateTime Serialization Support in aiokafka

[RebelRaider]

Describe the solution you’d like
I would like to request a feature that introduces automatic serialization and deserialization support for datetime objects in aiokafka. Currently, developers manually handle the conversion of datetime data to a string or timestamp format before sending it to Kafka. It would be great if aiokafka could natively support datetime objects by automatically serializing them into a standard format (e.g., ISO 8601 or timestamp) during message production, and deserializing them upon message consumption.

This feature would greatly reduce the manual overhead for handling date and time data, and also help ensure consistency in how datetime objects are represented across various Kafka topics and consumers.

Additional context
Other Python Kafka library examples:
The confluent-kafka-python library has a native support for date/time-based schemas when using Avro serialization.
The kafka-python library allows integration with custom serializers, which can be extended to support datetime objects, though it’s not automatic.

Thank you for considering this request, and please feel free to reach out if any further clarification or details are needed.

[ods]

aiokafka allows to supply custom serializer/deserialer as well. But this feature is considered for deprecation, since it stands in the way of adding type annotations.

[vmaurin]

aiokafka allows to supply custom serializer/deserialer as well. But this feature is considered for deprecation, since it stands in the way of adding type annotations.

Should it be deprecated though ? The type annotations could be resolved with a type parameter on the consumer/producer class.

On my own usage of aiokafka:

• sometime I am using serializer passed to the producer/consumer classes, sometime it is the caller handling it (for example when reusing a producer for multiple kind of topics/messages)
• I am using Pydantic, and from/to json, but it should be the caller to decide how to serialize/deserialize things
• the only "pain point" with date, are the timestamp_ms fields on record, where it could be nice to have helper/lazy property to move from datetime to timestamp_ms (but at the moment, we have these helpers in our code base and it is quite simple like int(dt.timestamp() * 1000)

[ods]

The type annotations could be resolved with a type parameter on the consumer/producer class.

This makes consumer and producer classes generic of 2 parameters, the types accepted by serializer/returned from deserializer for key and value. This doesn't look like a good design. But we need to find a better alternative first, so any ideas are welcome.

[vmaurin]

Yeah that could be a lot. In my company, we have built our own "kafka streams" based on aiokafka, and we do have things like this
def create_stateless_processor[InputKeyType, InputValueType, OutputKeyType, OutputValueType](
or event worse

def create_stateful_processor[
        InputKeyType,
        InputValueType,
        StateKeyType,
        StateValueType,
        OutputKeyType,
        OutputValueType,
    ](

It is also a bit similar in the java client (streams or regular)

• https://kafka.apache.org/38/javadoc/org/apache/kafka/streams/kstream/Consumed.html
• https://kafka.apache.org/38/javadoc/org/apache/kafka/clients/consumer/Consumer.html

In python, the type parameter is kind of verbose but on declaration only. On usage, we do have type checking without additional verbosity (the checker can "guess" the type according to the thing that are passed into the constructor for example)

So we could have something like this

def deserializer(serialized: bytes) -> dict[str, Any]:
    return json.loads(serialized)

async def consume():
    # consumer will decompress messages automatically
    # in accordance to compression type specified in producer
    consumer = AIOKafkaConsumer(
        'foobar',
        bootstrap_servers='localhost:9092',
        value_deserializer=deserializer,
        auto_offset_reset='earliest')
    await consumer.start()
    data = await consumer.getmany(timeout_ms=10000)
    for tp, messages in data.items():
        for message in messages:
            print(type(message.value), message.value)
    await consumer.stop()

asyncio.run(consume())

that would automatically infer the type of message.value from the deserializer method passed to the consumer constructor (but yeah, in the aiokafka codebase itself, you start having type parameters everywhere)

[ods]

Right now, when passing consumer (and similar for producer) around, you can declare it as AIOKafkaConsumer. But when types are added and we make it generic, it will be turned into something like AIOKafkaConsumer[bytes, bytes] or AIOKafkaConsumer[SomeKeyType, SomeValueType]. It becomes even more complicated when different topics require different (de)serializers. I believe it would be better to represent conversion with a separate entity, moving generic away from consumer/producer.

[vmaurin]

I guess then "wrapper" classes could be a solution ?

def deserializer(serialized: bytes) -> dict[str, Any]:
    return json.loads(serialized)

message_reader = MessageReader(value_deserializer=deserializer)

async def consume():
    # consumer will decompress messages automatically
    # in accordance to compression type specified in producer
    consumer = AIOKafkaConsumer(
        'foobar',
        bootstrap_servers='localhost:9092',
        auto_offset_reset='earliest')
    await consumer.start()
    data = await consumer.getmany(timeout_ms=10000)
    for tp, messages in message_reader.read(data.items()):
        for message in messages:
            print(type(message.value), message.value)
    await consumer.stop()

Not sure then what would be the benefit of having such class part of aiokafka directly

[ods]

I guess then "wrapper" classes could be a solution ?

Right, it's one of the possible solution. Another one could be to make generic methods that accept (de)serializers instead of making generic whole class.

As for me, I prefer to apply (de)serialization explicitly to each message: this way you it's clear how handle (de)serialization errors properly.