I am trying to consume a kafka topic from spring boot application. I am using Spring cloud stream with below mentioned version
- Spring-boot-starter-parent: 2.5.7
- Spring cloud version: 2020.0.4
Below are the code and configuration
application.yml
spring:
zipkin:
sender:
type: kafka
kafka:
bootstrap-servers:
- localhost:19091
cloud:
stream:
bindings:
audit-in-0:
destination: com.tonitingaurav.kafka.log
group: kafka-log-group
consumer:
concurrency: 10
max-attempts: 3
default-binder: kafka
kafka:
binder:
brokers:
- localhost:19091
Message Consumer Class
@Configuration
public class LogConsumer {
@Bean
Consumer<Log> audit(){
return log -> {
System.out.println(log.getMessage());
};
}
}
Below message publisher is publishing the messages properly. Publisher is written in different micro service.
@Component
public class LogEventPublisher {
@Autowired
@Qualifier(LogProducerKafkaConfig.KAFKA_LOG_PUBLISHER)
MessageChannel messageChannel;
public void logMessage(Log log) {
Message<Log> message = MessageBuilder.withPayload(log).build();
messageChannel.send(message);
}
}
pom.xml
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-binder-kafka-streams</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-stream-test-support</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.kafka</groupId>
<artifactId>spring-kafka-test</artifactId>
<scope>test</scope>
</dependency>
