{"id":17438,"date":"2023-10-08T08:12:47","date_gmt":"2023-10-08T07:12:47","guid":{"rendered":"https:\/\/www.baeldung.com\/?p=166771"},"modified":"2023-10-08T08:12:47","modified_gmt":"2023-10-08T07:12:47","slug":"how-to-subscribe-a-kafka-consumer-to-multiple-topics","status":"publish","type":"post","link":"https:\/\/gamefootballmobileanimeiphone.com\/index.php\/2023\/10\/08\/how-to-subscribe-a-kafka-consumer-to-multiple-topics\/","title":{"rendered":"How to Subscribe a Kafka Consumer to Multiple Topics"},"content":{"rendered":"

\"\"<\/p>\n

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

In this tutorial, we’ll learn how to subscribe a Kafka<\/a> consumer to multiple topics.<\/strong> This is a common requirement when the same business logic is used for various topics.<\/p>\n

2. Create Model Class<\/h2>\n
<\/div>\n

We’ll consider a simple payment system with two Kafka topics, one for card payments and the other for bank transfers. Let’s create the model class:<\/p>\n

public class PaymentData {\r\n    private String paymentReference;\r\n    private String type;\r\n    private BigDecimal amount;\r\n    private Currency currency;\r\n    \/\/ standard getters and setters\r\n}<\/code><\/pre>\n
3. Subscribe to Multiple Topics Using Kafka Consumer API<\/h2>\n
<\/div>\n
The first method we’ll discuss uses the Kafka Consumer API<\/a>. Let’s add the required Maven dependency:<\/a><\/p>\n
<dependency>\r\n    <groupId>org.apache.kafka<\/groupId>\r\n    <artifactId>kafka-clients<\/artifactId>\r\n    <version>3.5.1<\/version>\r\n<\/dependency><\/code><\/pre>\n
Let’s also configure the Kafka consumer:<\/p>\n
Properties properties = new Properties();\r\nproperties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KAFKA_CONTAINER.getBootstrapServers());\r\nproperties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());\r\nproperties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());\r\nproperties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");\r\nproperties.put(ConsumerConfig.GROUP_ID_CONFIG, "payments");\r\nkafkaConsumer = new KafkaConsumer<>(properties);<\/code><\/pre>\n
Before consuming messages, we need to subscribe kafkaConsumer<\/em> to both topics using the subscribe()<\/em> method:<\/strong><\/p>\n
kafkaConsumer.subscribe(Arrays.asList("card-payments", "bank-transfers"));<\/code><\/pre>\n
We’re ready now to test our configuration. Let’s publish one message on each of the topics:<\/p>\n
void publishMessages() throws Exception {\r\n    ProducerRecord<String, String> cardPayment = new ProducerRecord<>("card-payments", \r\n      "{\\"paymentReference\\":\\"A184028KM0013790\\", \\"type\\":\\"card\\", \\"amount\\":\\"275\\", \\"currency\\":\\"GBP\\"}");\r\n    kafkaProducer.send(cardPayment).get();\r\n    \r\n    ProducerRecord<String, String> bankTransfer = new ProducerRecord<>("bank-transfers",\r\n      "{\\"paymentReference\\":\\"19ae2-18mk73-009\\", \\"type\\":\\"bank\\", \\"amount\\":\\"150\\", \\"currency\\":\\"EUR\\"}");\r\n    kafkaProducer.send(bankTransfer).get();\r\n}<\/code><\/pre>\n
Finally, we can write the integration test:<\/p>\n
@Test\r\nvoid whenSendingMessagesOnTwoTopics_thenConsumerReceivesMessages() throws Exception {\r\n    publishMessages();\r\n    kafkaConsumer.subscribe(Arrays.asList("card-payments", "bank-transfers"));\r\n    int eventsProcessed = 0;\r\n    for (ConsumerRecord<String, String> record : kafkaConsumer.poll(Duration.ofSeconds(10))) {\r\n        log.info("Event on topic={}, payload={}", record.topic(), record.value());\r\n        eventsProcessed++;\r\n    }\r\n    assertThat(eventsProcessed).isEqualTo(2);\r\n}<\/code><\/pre>\n
4. Subscribe to Multiple Topics Using Spring Kafka<\/h2>\n
<\/div>\n
The second method we’ll discuss uses Spring Kafka<\/a>.<\/p>\n
Let’s add the spring-kafka<\/em><\/a> and jackson-databind<\/em><\/a> dependencies to our pom.xml<\/em>:<\/p>\n
<dependency> \r\n    <groupId>org.springframework.kafka<\/groupId> \r\n    <artifactId>spring-kafka<\/artifactId>\r\n    <version>3.0.11<\/version>\r\n<\/dependency>\r\n<dependency> \r\n    <groupId>com.fasterxml.jackson.core<\/groupId>\r\n    <artifactId>jackson-databind<\/artifactId>\r\n    <version>2.15.2<\/version>\r\n<\/dependency><\/code><\/pre>\n
Let’s also define the ConsumerFactory<\/em> and ConcurrentKafkaListenerContainerFactory<\/em> beans:<\/p>\n
@Bean\r\npublic ConsumerFactory<String, PaymentData> consumerFactory() {\r\n    List<String, String> config = new HashMap<>();\r\n    config.put(BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);\r\n    config.put(VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);\r\n    return new DefaultKafkaConsumerFactory<>(\r\n      config, new StringDeserializer(), new JsonDeserializer<>(PaymentData.class));\r\n}\r\n@Bean\r\npublic ConcurrentKafkaListenerContainerFactory<String, PaymentData> containerFactory() {\r\n    ConcurrentKafkaListenerContainerFactory<String, PaymentData> factory =\r\n      new ConcurrentKafkaListenerContainerFactory<>();\r\n    factory.setConsumerFactory(consumerFactory());\r\n    return factory;\r\n}<\/code><\/pre>\n
We need to subscribe to both topics using the topics<\/em> attribute of the @KafkaListener<\/em><\/a> annotation:<\/strong><\/p>\n
@KafkaListener(topics = { "card-payments", "bank-transfers" }, groupId = "payments")<\/code><\/pre>\n
Finally, we can create the consumer. Additionally, we’re also including the Kafka header to identify the topic where the message was received:<\/p>\n
@KafkaListener(topics = { "card-payments", "bank-transfers" }, groupId = "payments")\r\npublic void handlePaymentEvents(\r\n  PaymentData paymentData, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic) {\r\n    log.info("Event on topic={}, payload={}", topic, paymentData);\r\n}<\/code><\/pre>\n
Let’s validate our configuration<\/a>:<\/p>\n
@Test\r\npublic void whenSendingMessagesOnTwoTopics_thenConsumerReceivesMessages() throws Exception {\r\n    CountDownLatch countDownLatch = new CountDownLatch(2);\r\n    doAnswer(invocation -> {\r\n        countDownLatch.countDown();\r\n        return null;\r\n    }).when(paymentsConsumer)\r\n      .handlePaymentEvents(any(), any());\r\n    kafkaTemplate.send("card-payments", createCardPayment());\r\n    kafkaTemplate.send("bank-transfers", createBankTransfer());\r\n    assertThat(countDownLatch.await(5, TimeUnit.SECONDS)).isTrue();\r\n}<\/code><\/pre>\n
5. Subscribe to Multiple Topics Using Kafka CLI<\/h2>\n
<\/div>\n
Kafka CLI is the last method we’ll discuss.<\/p>\n
First, let’s send a message on each topic:<\/p>\n
$ bin\/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic card-payments\r\n>{"paymentReference":"A184028KM0013790", "type":"card", "amount":"275", "currency":"GBP"}\r\n$ bin\/kafka-console-producer.sh --bootstrap-server localhost:9092 --topic bank-transfers\r\n>{"paymentReference":"19ae2-18mk73-009", "type":"bank", "amount":"150", "currency":"EUR"}<\/code><\/pre>\n
Now, we can start the Kafka CLI consumer. The include<\/em> option allows us to specify the list of topics to include for message consumption:<\/strong><\/p>\n
$ bin\/kafka-console-consumer.sh --from-beginning --bootstrap-server localhost:9092 --include "card-payments|bank-transfers"\r\n<\/code><\/pre>\n
Here’s the output when we run the previous command:<\/p>\n
{"paymentReference":"A184028KM0013790", "type":"card", "amount":"275", "currency":"GBP"}\r\n{"paymentReference":"19ae2-18mk73-009", "type":"bank", "amount":"150", "currency":"EUR"}<\/code><\/pre>\n
6. Conclusion<\/h2>\n
<\/div>\n
In this article, we learned three different methods of subscribing a Kafka consumer to multiple topics.\u00a0This is useful when implementing the same functionality for several topics.<\/p>\n
The first two methods are based on Kafka Consumer API and Spring Kafka and can be integrated into an existing application. The last one uses Kafka CLI and can be used to verify multiple topics quickly.<\/p>\n
As always, the complete code can be found over on GitHub<\/a>.<\/p>\n
\"\"<\/p>\n
<\/a> <\/a> <\/a> <\/a> <\/a> <\/a> <\/a> <\/div>\n\n
Commercials Cooperation Advertisements:<\/b><\/h2>\r\n

(1) IT Teacher IT Freelance
 <\/p>\r\nIT\u96fb\u8166\u88dc\u7fd2<\/a>
\u7acb\u523b\u8a3b\u518a\u53ca\u5831\u540d\u96fb\u8166\u88dc\u7fd2\u8ab2\u7a0b\u5427! <\/a>
\r\n\r\n\u7535\u5b50\u8ba1\u7b97\u673a -\u6559\u80b2 -IT \u96fb\u8166\u73ed\u201d ( IT\u96fb\u8166\u88dc\u7fd2 ) \u63d0\u4f9b\u4e00\u500b\u65b9\u4fbf\u7684\u7535\u5b50\u8ba1\u7b97\u673a \u6559\u80b2\u5e73\u53f0, \u70ba\u5927\u5bb6\u914d\u5c0d\u4fe1\u606f\u6280\u672f, \u96fb\u8166 \u8001\u5e2b, IT freelance \u548c programming expert. \u8b93\u5927\u5bb6\u65b9\u4fbf\u5730\u5c31\u80fd\u627e\u5230\u5408\u9069\u7684\u96fb\u8166\u88dc\u7fd2, \u96fb\u8166\u73ed, \u5bb6\u6559, \u79c1\u4eba\u8001\u5e2b.  
\r\n\r\nWe are a education and information platform which you can find a IT private tutorial teacher or freelance. 
\r\n\r\nAlso we provide different information about information technology, Computer, programming, mobile, Android, apple, game, movie, anime, animation\u2026 \r\n<\/p>\n

(2) ITSec
 <\/p>https:\/\/itsec.vip\/\r\nSecure<\/a> \r\n

 \r\n
www.ITSec.vip<\/a> 
 
 \r\n
www.Sraa.com.hk<\/a> 
 
 \r\n
www.ITSec.hk<\/a> 
 
 \r\n
www.Penetrationtest.hk<\/a> 
 
 \r\n
www.ITSeceu.uk<\/a> 
 
 \r\nSecure Your Computers from Cyber Threats and mitigate risks with professional services to defend Hackers. 

\r\nITSec provide IT Security and Compliance Services, including IT Compliance Services, Risk Assessment, IT Audit, Security Assessment and Audit, ISO 27001 Consulting and Certification, GDPR Compliance Services, Privacy Impact Assessment (PIA), Penetration test, Ethical Hacking, Vulnerabilities scan, IT Consulting, Data Privacy Consulting, Data Protection Services, Information Security Consulting, Cyber Security Consulting, Network Security Audit, Security Awareness Training. \r\n

Contact us right away. 

Email (Prefer using email to contact us): 
SalesExecutive@ITSec.vip<\/p>","protected":false},"excerpt":{"rendered":"
\"\"<\/p>\n
Learn three different methods of subscribing a Kafka consumer to multiple topics.<\/p>\n
<\/a>\u00a0<\/a>\u00a0<\/a>\u00a0<\/a>\u00a0<\/a>\u00a0<\/a>\u00a0<\/a>\u00a0<\/div>\n","protected":false},"author":864,"featured_media":0,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"_monsterinsights_skip_tracking":false,"_monsterinsights_sitenote_active":false,"_monsterinsights_sitenote_note":"","_monsterinsights_sitenote_category":0,"site-container-style":"default","site-container-layout":"default","site-sidebar-layout":"default","disable-article-header":"default","disable-site-header":"default","disable-site-footer":"default","disable-content-area-spacing":"default","footnotes":""},"categories":[22],"tags":[61,122,127,129,124,128,125,132,131,133,126,130,123,66,94,88,97,56,64,65,60,112,40,75,95,104,33,120,7959,105,101,98,115,30,29,41,86,70,69,68,72,71,26,118,108,87,46,55,48,52,54,51,50,83,62,58,57,109,35,59,63,85,79,82,96,80,27,81,114,44,42,43,45,38,39,110,117,100,111,116,73,89,90,92,91,93,84,78,37,102,34,36,77,67,74,99,113,119,28,121,32,47,49,53,103,31,76],"class_list":["post-17438","post","type-post","status-publish","format-standard","hentry","category-mobile","tag-airpods","tag-anime","tag-anime-characters","tag-anime-cosplay","tag-anime-edits","tag-anime-merchandise","tag-anime-movies","tag-anime-news","tag-anime-recommendations","tag-anime-reviews","tag-anime-series","tag-anime-streaming","tag-animes","tag-app-store","tag-app-store-samsung","tag-appgallery","tag-appgallery-oneplus","tag-apple","tag-apple-music","tag-apple-tv","tag-apple-watch","tag-bbc-sport","tag-best-mobile-games","tag-bixby","tag-bixby-xiaomi","tag-champions-league","tag-cyberpunk","tag-cyberpunk-2077","tag-data","tag-fantasy-football","tag-fifa","tag-football","tag-formula-1","tag-fortnite","tag-free-fire","tag-free-mobile-games","tag-freebuds-pro","tag-galaxy-a52","tag-galaxy-note-20","tag-galaxy-s21","tag-galaxy-watch-4","tag-galaxy-z-fold-3","tag-game","tag-games","tag-golf","tag-harmonyos","tag-how-to-backup-iphone","tag-how-to-factory-reset-iphone","tag-how-to-reset-iphone","tag-how-to-restore-iphone","tag-how-to-unlock-iphone","tag-how-to-unlock-iphone-5","tag-how-to-unlock-iphone-6","tag-huawei","tag-ios","tag-ipad","tag-iphone","tag-live-soccer","tag-lol","tag-macbook","tag-macos","tag-mate-40-pro","tag-mi-11-lite","tag-mi-home-security-camera-basic-1080p","tag-mi-home-security-camera-basic-1080p-huawei","tag-mi-smart-band-6","tag-minecraft","tag-miui","tag-mlb-scores","tag-mobile-game-design","tag-mobile-game-development","tag-mobile-game-marketing","tag-mobile-game-monetization","tag-mobile-games","tag-mobile-gaming","tag-nba-scores","tag-nba-standings","tag-nfl","tag-nfl-scores","tag-nhl-scores","tag-one-ui","tag-oneplus","tag-oneplus-9-pro","tag-oneplus-buds-pro","tag-oneplus-nord-ce-5g","tag-oxygenos","tag-p40-pro-plus","tag-poco-x3-pro","tag-pokemon","tag-premier-league","tag-pubg","tag-pubg-mobile","tag-redmi-note-10-pro","tag-samsung","tag-samsung-pay","tag-soccer","tag-sports","tag-steam","tag-steeam","tag-top-10-anime","tag-valorant","tag-when-do-the-iphone-7-come-out","tag-when-does-the-iphone-7-come-out","tag-when-is-the-iphone-7-coming-out","tag-world-cup","tag-xbox-series-x","tag-xiaomi"],"aioseo_notices":[],"_links":{"self":[{"href":"https:\/\/gamefootballmobileanimeiphone.com\/index.php\/wp-json\/wp\/v2\/posts\/17438","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/gamefootballmobileanimeiphone.com\/index.php\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/gamefootballmobileanimeiphone.com\/index.php\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/gamefootballmobileanimeiphone.com\/index.php\/wp-json\/wp\/v2\/users\/864"}],"replies":[{"embeddable":true,"href":"https:\/\/gamefootballmobileanimeiphone.com\/index.php\/wp-json\/wp\/v2\/comments?post=17438"}],"version-history":[{"count":1,"href":"https:\/\/gamefootballmobileanimeiphone.com\/index.php\/wp-json\/wp\/v2\/posts\/17438\/revisions"}],"predecessor-version":[{"id":17439,"href":"https:\/\/gamefootballmobileanimeiphone.com\/index.php\/wp-json\/wp\/v2\/posts\/17438\/revisions\/17439"}],"wp:attachment":[{"href":"https:\/\/gamefootballmobileanimeiphone.com\/index.php\/wp-json\/wp\/v2\/media?parent=17438"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/gamefootballmobileanimeiphone.com\/index.php\/wp-json\/wp\/v2\/categories?post=17438"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/gamefootballmobileanimeiphone.com\/index.php\/wp-json\/wp\/v2\/tags?post=17438"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}