iSharkFly-Docs/java-tutorials
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #8782 from srzamfir/feature/BAEL-3185_Kafka_not_strating

Browse Source 
Feature/bael 3185 kafka not strating
This commit is contained in:
Eric Martin 2020-02-28 00:25:37 -06:00 committed by GitHub
commit 8bd02c32e3
4 changed files with 24 additions and 22 deletions
Show Stats 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

@ -108,8 +108,10 @@ public class KafkaApplication {
@Override @Override
public void onSuccess(SendResult<String, String> result) { public void onSuccess(SendResult<String, String> result) {
System.out.println("Sent message=[" + message + "] with offset=[" + result.getRecordMetadata().offset() + "]"); System.out.println("Sent message=[" + message + "] with offset=[" + result.getRecordMetadata()
.offset() + "]");
} }
@Override @Override
public void onFailure(Throwable ex) { public void onFailure(Throwable ex) {
System.out.println("Unable to send message=[" + message + "] due to : " + ex.getMessage()); System.out.println("Unable to send message=[" + message + "] due to : " + ex.getMessage());
@ -158,7 +160,7 @@ public class KafkaApplication {
latch.countDown(); latch.countDown();
} }
@KafkaListener(topicPartitions = @TopicPartition(topic = "${partitioned.topic.name}", partitions = { "0", "3" })) @KafkaListener(topicPartitions = @TopicPartition(topic = "${partitioned.topic.name}", partitions = { "0", "3" }), containerFactory = "partitionsKafkaListenerContainerFactory")
public void listenToParition(@Payload String message, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) { public void listenToParition(@Payload String message, @Header(KafkaHeaders.RECEIVED_PARTITION_ID) int partition) {
System.out.println("Received Message: " + message + " from partition: " + partition); System.out.println("Received Message: " + message + " from partition: " + partition);
this.partitionLatch.countDown(); this.partitionLatch.countDown();

8
spring-kafka/src/main/java/com/baeldung/spring/kafka/KafkaTopicConfig.java
View File

@ -37,21 +37,21 @@ public class KafkaTopicConfig {
@Bean @Bean
public NewTopic topic1() { public NewTopic topic1() {
return new NewTopic(topicName, 1, (short) 1); return new NewTopic(topicName, 1, (short) 1);
} }
@Bean @Bean
public NewTopic topic2() { public NewTopic topic2() {
return new NewTopic(partionedTopicName, 6, (short) 1); return new NewTopic(partionedTopicName, 6, (short) 1);
} }
@Bean @Bean
public NewTopic topic3() { public NewTopic topic3() {
return new NewTopic(filteredTopicName, 1, (short) 1); return new NewTopic(filteredTopicName, 1, (short) 1);
} }
@Bean @Bean
public NewTopic topic4() { public NewTopic topic4() {
return new NewTopic(greetingTopicName, 1, (short) 1); return new NewTopic(greetingTopicName, 1, (short) 1);
} }
} }