mindol1004/spring-boot-rest
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Update KafkaApplication.java

Browse Source 
Added partition offset to the listener so that it can start consuming the messages from the beginning of the topic.
This commit is contained in:
Muhammad Abdullah Azam Khan
2022-06-29 02:57:35 +04:00
committed by GitHub
parent 2f91a6f66f
commit 28801f8517
1 changed files with 5 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-kafka/src/main/java/com/baeldung/spring/kafka/KafkaApplication.java
View File

@@ -18,6 +18,7 @@ import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload; import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.util.concurrent.ListenableFuture; import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback; import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.kafka.annotation.PartitionOffset;
@SpringBootApplication @SpringBootApplication
public class KafkaApplication { public class KafkaApplication {
@@ -142,7 +143,10 @@ public class KafkaApplication {
private CountDownLatch greetingLatch = new CountDownLatch(1); private CountDownLatch greetingLatch = new CountDownLatch(1);
@KafkaListener(topics = "${message.topic.name}", groupId = "foo", containerFactory = "fooKafkaListenerContainerFactory") @KafkaListener(topicPartitions =
{ @TopicPartition(topic = "${message.topic.name}",
partitionOffsets = @PartitionOffset(partition = "0", initialOffset = "0"))},
topics = "${message.topic.name}", groupId = "foo", containerFactory = "fooKafkaListenerContainerFactory")
public void listenGroupFoo(String message) { public void listenGroupFoo(String message) {
System.out.println("Received Message in group 'foo': " + message); System.out.println("Received Message in group 'foo': " + message);
latch.countDown(); latch.countDown();