Incorrect Kafka message timestamps

[ChrisDev83]

Noticed that when I have my topics set to use LogAppendTime as the timestamp, the times I am getting back after consuming in k6 don't match up. I've checked this by using Kafka console consumer and akhq to view the message time. Both console consumer and akhq match but k6 has a different time. It's not a rounding issue either.

I have 2 topics with LogAppendTime configured. App consumes from TopicA and produces to TopicB but k6 timestamps show TopicB message being before TopicA.

Any ideas?

[mostafa]

Hey @ChrisDev83,

Thanks for reporting this. It's gonna be fixed in #372. Please test it and provide feedback.

[ChrisDev83]

@mostafa I think this is actually a kafka-go issue/feature. It seems to use the CreateTime rather than LogAppendTime somehow. I tested using a simply kafka-go consumer and compared to the console consumer from kafka tools. The console consumer shows the LogAppendTime for Topic B but the kafka-go consumer shows what looks like the time from the message on TopicA (its a kafka streams app that consumes from A and produces to B)

[mostafa]

@ChrisDev83 Interesting! I'll see if this is a configuration option in kafka-go, and I'll make it configurable via JS.

I just saw this: https://github.com/segmentio/kafka-go/discussions/1415

[mostafa]

I guess this is why: https://github.com/mostafa/xk6-kafka/blob/2f6cf2f10abaf0ed56889fc741f0c9a4aca7b51d/writer.go#L257-L261

We explicitly set the time to "now" if it is not set on the message via JS, which means that we probably bypass LogAppendTime. I can create a tiny PR, and would need your help testing it. WDYT?

[ChrisDev83]

I think the broker should still overwrite that value even if you set it?

Ref: https://developer.confluent.io/courses/kafka-streams/time-concepts/#log-append-time

[ChrisDev83]

Seems like for streams, Kafka leaves the message level timestamp as the timestamp from the original message (to avoid recompression or seomthing) and updates the batch level timestamp. Looks like kafka-go uses record level timestamp. Tricky one!

Looks like kafka-go doesnt expose the time for a batch.

https://pkg.go.dev/github.com/segmentio/kafka-go#Batch

[mostafa]

Have you tried adding message.timestamp.type=LogAppendTime (as ConfigEntry) to TopicConfig (in JS) while creating it? (I don't think it changes anything, though.)

[ChrisDev83]

The topics were created separately with that value set just like you have it.

[mostafa]

Then, what do you think/suspect the root cause is?

[ChrisDev83]

Had a bit of a look last night, I can't find concrete documentation but it seems to be some difference between the timestamp of a record vs timestamp of a batch. Console consumer and AKHQ, for example, match up in terms of the timestamp they show, but kafka-go shows the timestamp of the original message after being processed through kafka streams. I'll have another look when I get some time! Hard to pin point the root cause.

[github-actions[bot]]

This issue is stale because it has been open 30 days with no activity. Remove stale label or comment or this will be closed in 5 days.

[github-actions[bot]]

This issue was closed because it has been stalled for 5 days with no activity.