Spring Kafka - how to fetch timestamp (event time) when message was produced

Question

I have a requirement to fetch timestamp (event-time) when the message was produced, in the kafka consumer application. I am aware of the timestampExtractor, which can be used with kafka stream , but my requirement is different as I am not using stream to consume message.

My kafka producer is as follows :

@Override
public void run(ApplicationArguments args) throws Exception {


    List<String> names = Arrays.asList("priya", "dyser", "Ray", "Mark", "Oman", "Larry");
    List<String> pages = Arrays.asList("blog", "facebook", "instagram", "news", "youtube", "about");
    Runnable runnable = () -> {
        String rPage = pages.get(new Random().nextInt(pages.size()));
        String rName = pages.get(new Random().nextInt(names.size()));
        PageViewEvent pageViewEvent = new PageViewEvent(rName, rPage, Math.random() > .5 ? 10 : 1000);

        Message<PageViewEvent> message =  MessageBuilder
                .withPayload(pageViewEvent).
                setHeader(KafkaHeaders.MESSAGE_KEY, pageViewEvent.getUserId().getBytes())
                        .build();

        try {
            this.pageViewsOut.send(message);
            log.info("sent " + message);
        } catch (Exception e) {
            log.error(e);
        }
    };

Kafka Consumer is implemented using Spring kafka @KafkaListener.

@KafkaListener(topics = "test1" , groupId = "json", containerFactory = "kafkaListenerContainerFactory")

    public void receive(@Payload PageViewEvent data,@Headers MessageHeaders headers) {
        LOG.info("Message received");
        LOG.info("received data='{}'", data);
 }

Container factory configuration

   @Bean
   public ConsumerFactory<String,PageViewEvent > priceEventConsumerFactory() {

        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "json");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(), new JsonDeserializer<>(PageViewEvent.class));



    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, PageViewEvent> priceEventsKafkaListenerContainerFactory() {
        ConcurrentKafkaListenerContainerFactory<String, PageViewEvent> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(priceEventConsumerFactory());
        return factory;
    }

The producer, which is sending the message when I print give me below data :

[payload=PageViewEvent(userId=blog, page=about, duration=10), headers={id=8ebdad85-e2f7-958f-500e-4560ac0970e5, kafka_messageKey=[B@71975e1a, contentType=application/json, timestamp=1553041963803}]

This does have a produced timestamp. How can I fetch the message produced time stamp with Spring kafka?

Answer 1

RECEIVED_TIMESTAMP means it is the time stamp from the record that was received not the time it was received.. We avoid putting it in TIMESTAMP to avoid inadvertent propagation to an outbound message.

Answer 2

You can use something like below:

final Producer<String, String> producer = new KafkaProducer<String, String>(properties);
        long time = System.currentTimeMillis();
        final CountDownLatch countDownLatch = new CountDownLatch(5);
        int count=0;
        try {
            for (long index = time; index < time + 10; index++) {
                String key = null;
                count++;
                if(count<=5)
                    key = "id_"+ Integer.toString(1);
                else
                    key = "id_"+ Integer.toString(2);
                final ProducerRecord<String, String> record =
                        new ProducerRecord<>(TOPIC, key, "B2B Sample Message: " + count);
                producer.send(record, (metadata, exception) -> {
                    long elapsedTime = System.currentTimeMillis() - time;
                    if (metadata != null) {
                        System.out.printf("sent record(key=%s value=%s) " +
                                        "meta(partition=%d, offset=%d) time=%d timestamp=%d\n",
                                record.key(), record.value(), metadata.partition(),
                                metadata.offset(), elapsedTime, metadata.timestamp());
                        System.out.println("Timestamp:: "+metadata.timestamp() );
                    } else {
                        exception.printStackTrace();
                    }
                    countDownLatch.countDown();
                });
            }
            try {
                countDownLatch.await(25, TimeUnit.SECONDS);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }finally {
            producer.flush();
            producer.close();
        }

    }