{"id":329563,"date":"2024-05-24T22:38:06","date_gmt":"2024-05-24T22:38:06","guid":{"rendered":"https:\/\/www.baeldung.com\/?p=181972"},"modified":"2024-05-24T22:38:06","modified_gmt":"2024-05-24T22:38:06","slug":"consumer-processing-of-kafka-messages-with-delay","status":"publish","type":"post","link":"https:\/\/itteacheritfreelance.hk\/wordpress\/index.php\/2024\/05\/24\/consumer-processing-of-kafka-messages-with-delay\/","title":{"rendered":"Consumer Processing of Kafka Messages With Delay"},"content":{"rendered":"

<\/p>\n

\"start<\/p>\n

1. Overview<\/h2>\n
<\/div>\n

Apache Kafka<\/a> is an event streaming platform that collects, processes, stores, and integrates data at scale. Sometimes, we may want to delay the processing of messages from Kafka. An example is a customer order processing system designed to process orders after a delay of X seconds, accommodating cancellations within this timeframe.<\/p>\n

In this article, we’ll explore consumer processing of Kafka messages with delay using Spring Kafka<\/a>. Although Kafka doesn’t provide out-of-the-box support for the delayed consumption of messages, we’ll look at an alternative option for implementation.<\/p>\n

2. Application Context<\/h2>\n
<\/div>\n

Kafka offers multiple ways to retry on errors. We’ll use this retry mechanism to delay the consumer processing of messages. Therefore, it’s worth understanding how Kafka retry<\/a> works.<\/strong><\/p>\n

Let’s consider an order processing application where a customer can place an order on a UI. The user can cancel mistakenly placed orders within 10 seconds. These orders go to the Kafka topic web.orders<\/em>, where our application processes them.<\/p>\n

An external service exposes the latest order status (CREATED, ORDER_CONFIRMED, ORDER_PROCESSED, DELETED<\/em>). Our application needs to receive the message, wait for 10 seconds, and check with the external service to process the order if it’s in CONFIRMED<\/em> status, i.e., the user hasn’t canceled it within the 10 seconds.<\/p>\n

For testing, the internal orders received from web.orders.internal<\/em> shouldn’t be delayed.<\/p>\n

Let’s add a simple Order<\/em> model that has orderGeneratedDateTime<\/em> populated by the producer and orderProcessedTime<\/em> populated by the consumer after a delayed duration:<\/p>\n

@Data\r\n@Builder\r\n@AllArgsConstructor\r\n@NoArgsConstructor\r\npublic class Order {\r\n    private UUID orderId;\r\n    private LocalDateTime orderGeneratedDateTime;\r\n    private LocalDateTime orderProcessedTime;\r\n    private List<String> address;\r\n    private double price;\r\n}<\/code><\/pre>\n
3. Kafka Listener and External Service<\/h2>\n
<\/div>\n
Next, we’ll add a listener for topic consumption and a service that exposes the status of the orders.<\/strong><\/p>\n
Let’s add a KafkaListener<\/a><\/em> which reads and processes messages from topics web.orders<\/em> and web.internal.orders<\/em> :<\/p>\n
@RetryableTopic(attempts = "1", include = KafkaBackoffException.class, dltStrategy = DltStrategy.NO_DLT)\r\n@KafkaListener(topics = { "web.orders", "web.internal.orders" }, groupId = "orders")\r\npublic void handleOrders(String order) throws JsonProcessingException {\r\n    Order orderDetails = objectMapper.readValue(order, Order.class);\r\n    OrderService.Status orderStatus = orderService.findStatusById(orderDetails.getOrderId());\r\n    if (orderStatus.equals(OrderService.Status.ORDER_CONFIRMED)) {\r\n        orderService.processOrder(orderDetails);\r\n    }\r\n}<\/code><\/pre>\n
It’s important to include KafkaBackoffException<\/em> so that the listener allows retries. For simplicity, let’s consider that the external OrderService<\/em> always returns the order status as CONFIRMED. <\/em>Also, the processOrder()<\/em> method sets the order processed time as the current time and saves the order into a HashMap<\/em>:<\/p>\n
@Service\r\npublic class OrderService {\r\n    HashMap<UUID, Order> orders = new HashMap<>();\r\n    public Status findStatusById(UUID orderId) {\r\n        return Status.ORDER_CONFIRMED;\r\n    }\r\n    public void processOrder(Order order) {\r\n        order.setOrderProcessedTime(LocalDateTime.now());\r\n        orders.put(order.getOrderId(), order);\r\n    }\r\n}<\/code><\/pre>\n
4. Custom Delayed Message Listener<\/h2>\n
<\/div>\n
Spring-Kafka comes up with the KafkaBackoffAwareMessageListenerAdapter<\/em><\/a> which extends AbstractAdaptableMessageListener<\/em> and implements AcknowledgingConsumerAwareMessageListener<\/em>. This adapter examines the backoff dueTimestamp<\/em> header and either backs off the message by invoking KafkaConsumerBackoffManager<\/em> or retries the processing.<\/strong><\/p>\n
Let’s now implement the DelayedMessageListenerAdapter <\/em>similar to KafkaBackoffAwareMessageListenerAdapter<\/em>. This adapter should provide flexibility to configure delays per topic along with a default delay of 0<\/em> seconds:<\/p>\n
public class DelayedMessageListenerAdapter<K, V> extends AbstractDelegatingMessageListenerAdapter<MessageListener<K, V>> \r\n  implements AcknowledgingConsumerAwareMessageListener<K, V> {\r\n    \/\/ Field declaration and constructor\r\n    public void setDelayForTopic(String topic, Duration delay) {\r\n        Objects.requireNonNull(topic, "Topic cannot be null");\r\n        Objects.requireNonNull(delay, "Delay cannot be null");\r\n        this.logger.debug(() -> String.format("Setting delay %s for listener id %s", delay, this.listenerId));\r\n        this.delaysPerTopic.put(topic, delay);\r\n    }\r\n    public void setDefaultDelay(Duration delay) {\r\n        Objects.requireNonNull(delay, "Delay cannot be null");\r\n        this.logger.debug(() -> String.format("Setting delay %s for listener id %s", delay, this.listenerId));\r\n        this.defaultDelay = delay;\r\n    }\r\n    @Override\r\n    public void onMessage(ConsumerRecord<K, V> consumerRecord, Acknowledgment acknowledgment, Consumer<?, ?> consumer) throws KafkaBackoffException {\r\n        this.kafkaConsumerBackoffManager.backOffIfNecessary(createContext(consumerRecord,\r\n          consumerRecord.timestamp() + delaysPerTopic.getOrDefault(consumerRecord.topic(), this.defaultDelay)\r\n          .toMillis(), consumer));\r\n        invokeDelegateOnMessage(consumerRecord, acknowledgment, consumer);\r\n    }\r\n    private KafkaConsumerBackoffManager.Context createContext(ConsumerRecord<K, V> data, long nextExecutionTimestamp, Consumer<?, ?> consumer) {\r\n        return this.kafkaConsumerBackoffManager.createContext(nextExecutionTimestamp, \r\n          this.listenerId, \r\n          new TopicPartition(data.topic(), data.partition()), consumer);\r\n    }\r\n}<\/code><\/pre>\n
For every incoming message, this adapter first receives the record and checks the delay set for the topic. This will be set in the configuration, and if not set, it uses the default delay.<\/p>\n
The existing implementation of KafkaConsumerBackoffManager#backOffIfNecessary <\/em>method checks the difference between the context record timestamp and the current timestamp.\u00a0If the difference is positive, showing no consumption due, the partition pauses and raises a KafkaBackoffException<\/em><\/a>. Otherwise, it sends the record to the KafkaListener<\/em> method for consumption.<\/p>\n
5. Listener Configuration<\/h2>\n
<\/div>\n
The ConcurrentKafkaListenerContainerFactory<\/a><\/em> is the default implementation of Spring Kafka which is responsible for building containers for KafkaListener<\/em>.<\/strong> It allows us to configure the number of concurrent KafkaListener<\/em> instances. Each container can be considered a logical thread pool, where each thread is responsible for listening to messages from one or more Kafka topics.<\/p>\n
The DelayedMessageListenerAdapter<\/em> needs to be configured with the listener by declaring a custom ConcurrentKafkaListenerContainerFactory. <\/em>We can set the delay for specific topics like web.orders<\/em> and also set a default delay of 0<\/em> for any other topics:<\/p>\n
@Bean\r\npublic ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainerFactory(ConsumerFactory<Object, Object> consumerFactory, \r\n  ListenerContainerRegistry registry, TaskScheduler scheduler) {\r\n    ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();\r\n    factory.setConsumerFactory(consumerFactory);\r\n    KafkaConsumerBackoffManager backOffManager = createBackOffManager(registry, scheduler);\r\n    factory.getContainerProperties()\r\n      .setAckMode(ContainerProperties.AckMode.RECORD);\r\n    factory.setContainerCustomizer(container -> {\r\n        DelayedMessageListenerAdapter<Object, Object> delayedAdapter = wrapWithDelayedMessageListenerAdapter(backOffManager, container);\r\n        delayedAdapter.setDelayForTopic("web.orders", Duration.ofSeconds(10));\r\n        delayedAdapter.setDefaultDelay(Duration.ZERO);\r\n        container.setupMessageListener(delayedAdapter);\r\n    });\r\n    return factory;\r\n}\r\n@SuppressWarnings("unchecked")\r\nprivate DelayedMessageListenerAdapter<Object, Object> wrapWithDelayedMessageListenerAdapter(KafkaConsumerBackoffManager backOffManager, \r\n  ConcurrentMessageListenerContainer<Object, Object> container) {\r\n    return new DelayedMessageListenerAdapter<>((MessageListener<Object, Object>) container.getContainerProperties()\r\n      .getMessageListener(), backOffManager, container.getListenerId());\r\n}\r\nprivate ContainerPartitionPausingBackOffManager createBackOffManager(ListenerContainerRegistry registry, TaskScheduler scheduler) {\r\n    return new ContainerPartitionPausingBackOffManager(registry, \r\n      new ContainerPausingBackOffHandler(new ListenerContainerPauseService(registry, scheduler)));\r\n}<\/code><\/pre>\n
Notably, setting the acknowledgment mode at the RECORD<\/em> level is essential to ensure that the consumer redelivers messages if an error happens during processing.<\/p>\n
Finally, we need to define a TaskScheduler<\/em> bean to resume paused partitions after the delay duration and this scheduler needs to be injected into the BackOffManager <\/em>which will be used by DelayedMessageListenerAdapter<\/em>:<\/p>\n
@Bean\r\npublic TaskScheduler taskScheduler() {\r\n    return new ThreadPoolTaskScheduler();\r\n}<\/code><\/pre>\n
6. Testing<\/h2>\n
<\/div>\n
Let’s ensure orders on the web.orders<\/em> topic undergo a 10-second delay before processing through testing:<\/p>\n
@Test\r\nvoid givenKafkaBrokerExists_whenCreateOrderIsReceived_thenMessageShouldBeDelayed() throws Exception {\r\n    \/\/ Given\r\n    var orderId = UUID.randomUUID();\r\n    Order order = Order.builder()\r\n      .orderId(orderId)\r\n      .price(1.0)\r\n      .orderGeneratedDateTime(LocalDateTime.now())\r\n      .address(List.of("41 Felix Avenue, Luton"))\r\n      .build();\r\n    String orderString = objectMapper.writeValueAsString(order);\r\n    ProducerRecord<String, String> record = new ProducerRecord<>("web.orders", orderString);\r\n    \r\n    \/\/ When\r\n    testKafkaProducer.send(record)\r\n      .get();\r\n    await().atMost(Duration.ofSeconds(1800))\r\n      .until(() -> {\r\n          \/\/ then\r\n          Map<UUID, Order> orders = orderService.getOrders();\r\n          return orders != null && orders.get(orderId) != null && Duration.between(orders.get(orderId)\r\n              .getOrderGeneratedDateTime(), orders.get(orderId)\r\n              .getOrderProcessedTime())\r\n            .getSeconds() >= 10;\r\n      });\r\n}<\/code><\/pre>\n
Next, we’ll test any orders to web.internal.orders<\/em> follow a default delay of 0<\/em> seconds:<\/p>\n
@Test\r\nvoid givenKafkaBrokerExists_whenCreateOrderIsReceivedForOtherTopics_thenMessageShouldNotBeDelayed() throws Exception {\r\n    \/\/ Given\r\n    var orderId = UUID.randomUUID();\r\n    Order order = Order.builder()\r\n      .orderId(orderId)\r\n      .price(1.0)\r\n      .orderGeneratedDateTime(LocalDateTime.now())\r\n      .address(List.of("41 Felix Avenue, Luton"))\r\n      .build();\r\n    String orderString = objectMapper.writeValueAsString(order);\r\n    ProducerRecord<String, String> record = new ProducerRecord<>("web.internal.orders", orderString);\r\n    \r\n    \/\/ When\r\n    testKafkaProducer.send(record)\r\n      .get();\r\n    await().atMost(Duration.ofSeconds(1800))\r\n      .until(() -> {\r\n          \/\/ Then\r\n          Map<UUID, Order> orders = orderService.getOrders();\r\n          System.out.println("Time...." + Duration.between(orders.get(orderId)\r\n              .getOrderGeneratedDateTime(), orders.get(orderId)\r\n              .getOrderProcessedTime())\r\n            .getSeconds());\r\n          return orders != null && orders.get(orderId) != null && Duration.between(orders.get(orderId)\r\n              .getOrderGeneratedDateTime(), orders.get(orderId)\r\n              .getOrderProcessedTime())\r\n            .getSeconds() <= 1;\r\n      });\r\n}<\/code><\/pre>\n
7. Conclusion<\/h2>\n
<\/div>\n
In this tutorial, we explored how a Kafka consumer can delay processing messages by fixed intervals.<\/p>\n
We can modify the implementation to dynamically set processing delays by utilizing embedded message durations as part of the message.<\/p>\n
As always, the source code for the examples is available over on GitHub<\/a>.<\/p>\n
\"\"<\/p>\n
<\/a> <\/a> <\/a> <\/a> <\/a> <\/a> <\/a> <\/div>\n\n
\\\"IT\u96fb\u8166\u88dc\u7fd2
\u7acb\u523b\u8a3b\u518a\u53ca\u5831\u540d\u96fb\u8166\u88dc\u7fd2\u8ab2\u7a0b\u5427!<\/figcaption><\/figure>\r\n<\/br>Find A Teacher Form:\r\n<\/br>https:\/\/docs.google.com\/forms\/d\/1vREBnX5n262umf4wU5U2pyTwvk9O-JrAgblA-wH9GFQ\/viewform?edit_requested=true#responses\r\n<\/br><\/br>Email:\r\n<\/br>public1989two@gmail.com






\r\nwww.itsec.hk<\/a>
\r\nwww.itsec.vip<\/a>
\r\nwww.itseceu.uk<\/a>
<\/p>","protected":false},"excerpt":{"rendered":"
\"start<\/p>\n
Explore consumer processing of Kafka messages with delay using Spring Kafka.<\/p>\n
<\/a>\u00a0<\/a>\u00a0<\/a>\u00a0<\/a>\u00a0<\/a>\u00a0<\/a>\u00a0<\/a>\u00a0<\/div>\n<\/div>","protected":false},"author":1979,"featured_media":0,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"slim_seo":{"title":"Consumer Processing of Kafka Messages With Delay - ITTeacherITFreelance.hk","description":"Explore consumer processing of Kafka messages with delay using Spring Kafka. \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0 \u00a0"},"footnotes":""},"categories":[6,1307],"tags":[10803],"_links":{"self":[{"href":"https:\/\/itteacheritfreelance.hk\/wordpress\/index.php\/wp-json\/wp\/v2\/posts\/329563"}],"collection":[{"href":"https:\/\/itteacheritfreelance.hk\/wordpress\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/itteacheritfreelance.hk\/wordpress\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/itteacheritfreelance.hk\/wordpress\/index.php\/wp-json\/wp\/v2\/users\/1979"}],"replies":[{"embeddable":true,"href":"https:\/\/itteacheritfreelance.hk\/wordpress\/index.php\/wp-json\/wp\/v2\/comments?post=329563"}],"version-history":[{"count":1,"href":"https:\/\/itteacheritfreelance.hk\/wordpress\/index.php\/wp-json\/wp\/v2\/posts\/329563\/revisions"}],"predecessor-version":[{"id":329564,"href":"https:\/\/itteacheritfreelance.hk\/wordpress\/index.php\/wp-json\/wp\/v2\/posts\/329563\/revisions\/329564"}],"wp:attachment":[{"href":"https:\/\/itteacheritfreelance.hk\/wordpress\/index.php\/wp-json\/wp\/v2\/media?parent=329563"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/itteacheritfreelance.hk\/wordpress\/index.php\/wp-json\/wp\/v2\/categories?post=329563"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/itteacheritfreelance.hk\/wordpress\/index.php\/wp-json\/wp\/v2\/tags?post=329563"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}