Handle timestamp value in response from Bridge

[im-konge]

After strimzi/strimzi-kafka-bridge#915 was added into Bridge, we need to handle it in our ConsumerRecord for HTTP response -> because without that, we are getting:

2024-08-01 16:24:45.758.12287383472543 �[1;31mERROR�[m HttpConsumerClient:175 - Failed to consume message due to: Unrecognized field "timestamp" (class io.strimzi.common.records.http.consumer.ConsumerRecord), not marked as ignorable (5 known properties: "value", "partition", "offset", "topic", "key"])
 at [Source: REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled); line: 1, column: 130] (through reference chain: java.lang.Object[][0]->io.strimzi.common.records.http.consumer.ConsumerRecord["timestamp"])
com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException: Unrecognized field "timestamp" (class io.strimzi.common.records.http.consumer.ConsumerRecord), not marked as ignorable (5 known properties: "value", "partition", "offset", "topic", "key"])
 at [Source: REDACTED (`StreamReadFeature.INCLUDE_SOURCE_IN_LOCATION` disabled); line: 1, column: 130] (through reference chain: java.lang.Object[][0]->io.strimzi.common.records.http.consumer.ConsumerRecord["timestamp"])
	at com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException.from(UnrecognizedPropertyException.java:61)
	at com.fasterxml.jackson.databind.DeserializationContext.handleUnknownProperty(DeserializationContext.java:1153)
	at com.fasterxml.jackson.databind.deser.std.StdDeserializer.handleUnknownProperty(StdDeserializer.java:2224)
	at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.handleUnknownProperty(BeanDeserializerBase.java:1793)
	at com.fasterxml.jackson.databind.deser.BeanDeserializerBase.handleUnknownVanilla(BeanDeserializerBase.java:1771)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.vanillaDeserialize(BeanDeserializer.java:316)
	at com.fasterxml.jackson.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:177)
	at com.fasterxml.jackson.databind.deser.std.ObjectArrayDeserializer.deserialize(ObjectArrayDeserializer.java:217)
	at com.fasterxml.jackson.databind.deser.std.ObjectArrayDeserializer.deserialize(ObjectArrayDeserializer.java:27)
	at com.fasterxml.jackson.databind.deser.DefaultDeserializationContext.readRootValue(DefaultDeserializationContext.java:342)
	at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:4899)
	at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3846)
	at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:3814)
	at io.strimzi.common.records.http.consumer.ConsumerRecordUtils.parseConsumerRecordsFromJson(ConsumerRecordUtils.java:22)
	at io.strimzi.http.consumer.HttpConsumerClient.consumeMessages(HttpConsumerClient.java:168)
	at io.strimzi.http.consumer.HttpConsumerClient.checkAndReceiveMessages(HttpConsumerClient.java:96)
	at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539)
	at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305)
	at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:840)

[ppatierno]

Looking at the error message, I am assuming the test is failing because producer is setting the timestamp but consumer is not able to decode it, right? Because timestamp is optional not mandatory for the bridge.

[im-konge]

@ppatierno
From what I tested, the timestamp was there every time.
In our Kafka producer, we are not setting the timestamp:

        return new ProducerRecord(configuration.getTopicName(), null, null, configuration.getMessageKey(),
            message, configuration.getHeaders());

where the second null should be timestamp.

For Bridge producer we are not setting the timestamp neither:

String record = "{\"records\":[{\"key\":\"key-" + numOfMessage + "\",\"value\":" + message + "}]}";

Is it optional? I thought that it will be there always (but that's something I will need to test).

[ppatierno]

@im-konge yeah sorry, I was stupid. Of course, on the consumer side it's always there. Even when the producer doesn't send it, the record will get the processed timestamp when it's ingested on Kafka and returned to the consumer.

[ppatierno]

LGTM.

[im-konge]

@ppatierno yeah I was just testing it and I bumped into article about it. So yes, it will be there always.
Thanks for checking it :)