1010import org .springframework .context .ConfigurableApplicationContext ;
1111import org .springframework .context .annotation .Bean ;
1212import org .springframework .kafka .annotation .KafkaListener ;
13+ import org .springframework .kafka .annotation .TopicPartition ;
1314import org .springframework .kafka .core .KafkaTemplate ;
15+ import org .springframework .kafka .support .KafkaHeaders ;
16+ import org .springframework .messaging .handler .annotation .Header ;
17+ import org .springframework .messaging .handler .annotation .Payload ;
1418
1519@ SpringBootApplication
1620public class KafkaApplication {
1721
1822 public static void main (String [] args ) throws Exception {
23+
1924 ConfigurableApplicationContext context = SpringApplication .run (KafkaApplication .class , args );
25+
2026 MessageProducer producer = context .getBean (MessageProducer .class );
27+ MessageListener listener = context .getBean (MessageListener .class );
28+ /*
29+ * Sending a Hello World message to topic 'baeldung'.
30+ * Must be recieved by both listeners with group foo
31+ * and bar with containerFactory fooKafkaListenerContainerFactory
32+ * and barKafkaListenerContainerFactory respectively.
33+ * It will also be recieved by the listener with
34+ * headersKafkaListenerContainerFactory as container factory
35+ */
2136 producer .sendMessage ("Hello, World!" );
37+ listener .latch .await (10 , TimeUnit .SECONDS );
38+
39+ /*
40+ * Sending message to a topic with 5 partition,
41+ * each message to a different partition. But as per
42+ * listener configuration, only the messages from
43+ * partition 0 and 3 will be consumed.
44+ */
45+ for (int i = 0 ; i < 5 ; i ++) {
46+ producer .sendMessageToPartion ("Hello To Partioned Topic!" , i );
47+ }
48+ listener .partitionLatch .await (10 , TimeUnit .SECONDS );
49+
50+ /*
51+ * Sending message to 'filtered' topic. As per listener
52+ * configuration, all messages with char sequence
53+ * 'World' will be discarded.
54+ */
55+ producer .sendMessageToFiltered ("Hello Baeldung!" );
56+ producer .sendMessageToFiltered ("Hello World!" );
57+ listener .filterLatch .await (10 , TimeUnit .SECONDS );
58+
59+ /*
60+ * Sending message to 'greeting' topic. This will send
61+ * and recieved a java object with the help of
62+ * greetingKafkaListenerContainerFactory.
63+ */
64+ producer .sendGreetingMessage (new Greeting ("Greetings" , "World!" ));
65+ listener .greetingLatch .await (10 , TimeUnit .SECONDS );
2266
23- MessageListener listener = context .getBean (MessageListener .class );
24- listener .latch .await (20 , TimeUnit .SECONDS );
25- Thread .sleep (60000 );
2667 context .close ();
27-
2868 }
2969
3070 @ Bean
@@ -42,18 +82,47 @@ public static class MessageProducer {
4282 @ Autowired
4383 private KafkaTemplate <String , String > kafkaTemplate ;
4484
85+ @ Autowired
86+ private KafkaTemplate <String , Greeting > greetingKafkaTemplate ;
87+
4588 @ Value (value = "${message.topic.name}" )
4689 private String topicName ;
4790
91+ @ Value (value = "${partitioned.topic.name}" )
92+ private String partionedTopicName ;
93+
94+ @ Value (value = "${filtered.topic.name}" )
95+ private String filteredTopicName ;
96+
97+ @ Value (value = "${greeting.topic.name}" )
98+ private String greetingTopicName ;
99+
48100 public void sendMessage (String message ) {
49101 kafkaTemplate .send (topicName , message );
50102 }
51103
104+ public void sendMessageToPartion (String message , int partition ) {
105+ kafkaTemplate .send (partionedTopicName , partition , message );
106+ }
107+
108+ public void sendMessageToFiltered (String message ) {
109+ kafkaTemplate .send (filteredTopicName , message );
110+ }
111+
112+ public void sendGreetingMessage (Greeting greeting ) {
113+ greetingKafkaTemplate .send (greetingTopicName , greeting );
114+ }
52115 }
53116
54117 public static class MessageListener {
55118
56- private CountDownLatch latch = new CountDownLatch (2 );
119+ private CountDownLatch latch = new CountDownLatch (3 );
120+
121+ private CountDownLatch partitionLatch = new CountDownLatch (2 );
122+
123+ private CountDownLatch filterLatch = new CountDownLatch (2 );
124+
125+ private CountDownLatch greetingLatch = new CountDownLatch (1 );
57126
58127 @ KafkaListener (topics = "${message.topic.name}" , group = "foo" , containerFactory = "fooKafkaListenerContainerFactory" )
59128 public void listenGroupFoo (String message ) {
@@ -67,6 +136,30 @@ public void listenGroupBar(String message) {
67136 latch .countDown ();
68137 }
69138
139+ @ KafkaListener (topics = "${message.topic.name}" , containerFactory = "headersKafkaListenerContainerFactory" )
140+ public void listenWithHeaders (@ Payload String message , @ Header (KafkaHeaders .RECEIVED_PARTITION_ID ) int partition ) {
141+ System .out .println ("Received Messasge: " + message + " from partition: " + partition );
142+ latch .countDown ();
143+ }
144+
145+ @ KafkaListener (topicPartitions = @ TopicPartition (topic = "${partitioned.topic.name}" , partitions = { "0" , "3" }))
146+ public void listenToParition (@ Payload String message , @ Header (KafkaHeaders .RECEIVED_PARTITION_ID ) int partition ) {
147+ System .out .println ("Received Message: " + message + " from partition: " + partition );
148+ this .partitionLatch .countDown ();
149+ }
150+
151+ @ KafkaListener (topics = "${filtered.topic.name}" , containerFactory = "filterKafkaListenerContainerFactory" )
152+ public void listenWithFilter (String message ) {
153+ System .out .println ("Recieved Message in filtered listener: " + message );
154+ this .filterLatch .countDown ();
155+ }
156+
157+ @ KafkaListener (topics = "${greeting.topic.name}" , containerFactory = "greetingKafkaListenerContainerFactory" )
158+ public void greetingListener (Greeting greeting ) {
159+ System .out .println ("Recieved greeting message: " + greeting );
160+ this .greetingLatch .countDown ();
161+ }
162+
70163 }
71164
72165}
0 commit comments