CODEWITHSUNDEEP
spring-bootapache-kafkaspring-cloud-streamspring-cloud-stream-binder-kafka
user3082820
user3082820

Reputation: 101

Spring Cloud Stream Consumer is not reading messages from KAFKA topic

I am trying to consume a kafka topic from spring boot application. I am using Spring cloud stream with below mentioned version

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>

Upvotes: 2

Views: 2468

Answers (1)

Oleg Zhurakousky
Oleg Zhurakousky

Reputation: 6126

You already posted a very similar question here and the response with two different solutions was provided. Also, here are the samples you can use as starting point - https://github.com/spring-cloud/spring-cloud-stream-samples

Upvotes: -1

Related Questions