Missing Timestamps

[ppbpdx]

I am using aiokafka==0.11.0 and have a very basic consumer setup. What I am noticing is that sometime messages are missing the timestamps. I have verified that the server is configured to set them by default, and, when I look in Kafdrop at the same message at the same offset I can see a timestamp there.

This issue seems to come and go without any change to the configuration.

Here is how I am setting up the consumer:

      consumer = AIOKafkaConsumer(
            *topics,
            bootstrap_servers=kafka_server,
            enable_auto_commit=True,
            auto_offset_reset="earliest",
            group_id=f"my-consumer{os.getpid()}",
            session_timeout_ms=1000 * 60,
        )

And here is where I am logging the errors when there is no timestamp:

message = await asyncio.wait_for(message_queue.get(), timeout=timeout_seconds)
                message_queue.task_done()

                message_value = message.value
                message_topic = message.topic

                if message_value:
                        timestamp = message.timestamp
                        if not timestamp:
                            logger.error('EntityStatistics kafka message is missing timestamp')

Any insight as to why this might be happening is very much appreciated.