CODEWITHSUNDEEP
kafka-consumer-apispring-kafka
Sabya
Sabya

Reputation: 207

Kafka Listener Method is not invoked. Consumer not consuming.

the aforementioned kafka consumer that should consume from a single topic. I can't use spring boot as am integrating kafka consumer api with a spring core web app..

spring xml configuration is as follows

<bean id="kafkaConsumerProperties" class="com.azuga.kafka.listeners.KafkaConsumerProperties">
    <constructor-arg type="java.lang.String" value="127.0.0.1:9092" />
    <constructor-arg type="java.lang.String" value="tdm-group" />
    <constructor-arg type="java.lang.String" value="dbStreamer.azuga.tripDriverMapping" />
</bean>
<bean id="kafkaListenerConfig" class="com.azuga.kafka.listeners.KafkaListenerConfig">
    <property name="kafkaConsumerProperties" ref="kafkaConsumerProperties" />
</bean>
<bean id="kafkaContainerFactory" class="com.azuga.kafka.listeners.KafkaListenerContainerFactory"
    factory-method="kafkaContainerFactory">
</bean>

This is the class that creates the ListenerContainerFactory

@EnableKafka
public class KafkaListenerContainerFactory {

public static ConcurrentKafkaListenerContainerFactory<String, String> kafkaContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConcurrency(1);
    factory.setConsumerFactory(consumerFactory());
    factory.getContainerProperties().setPollTimeout(3000);
    return factory;
}

@SuppressWarnings("unchecked")
public static ConsumerFactory<String, String> consumerFactory() {
    return new DefaultKafkaConsumerFactory<>(KafkaListenerConfig.consumerProps(),
            KafkaListenerConfig.stringKeyDeserializer(), KafkaListenerConfig.stringKeyDeserializer());
}

}

This is my Listener Class annotated with @KafkaListener

package com.azuga.kafka.listeners;

import org.springframework.kafka.annotation.KafkaListener;
public class Listener {

@KafkaListener(topics = "dbStreamer.azuga.tripDriverMapping")
public void onMessage(String message) {
    System.out.println(message.toString());
}
}

This is the KafkaListenerConfig class which takes in the bootstrap servers, topic names etc.

@EnableKafka
public class KafkaListenerConfig {

private static KafkaConsumerProperties kafkaConsumerProperties;

public void setKafkaConsumerProperties(KafkaConsumerProperties kafkaConsumerProperties) {
    this.kafkaConsumerProperties = kafkaConsumerProperties;
}

public static Map<String, Object> consumerProps() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConsumerProperties.getBootstrap());
    props.put(ConsumerConfig.GROUP_ID_CONFIG, kafkaConsumerProperties.getGroup());
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
    props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "100");
    props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "15000");
    return props;
}

public static Deserializer stringKeyDeserializer() {
    return new StringDeserializer();
}

}

Upvotes: 2

Views: 2088

Answers (1)

Artem Bilan
Artem Bilan

Reputation: 121552

You have a bit unusual configuration for your application.

However I assume you are missing the fact that @EnableKafka is about @Configuration class. Therefore, according Spring Framework documentation, you have to use AnnotationConfigWebApplicationContext class:

* {@link org.springframework.web.context.WebApplicationContext WebApplicationContext}
 * implementation which accepts annotated classes as input - in particular
 * {@link org.springframework.context.annotation.Configuration @Configuration}-annotated
 * classes, but also plain {@link org.springframework.stereotype.Component @Component}
 * classes and JSR-330 compliant classes using {@code javax.inject} annotations. Allows
 * for registering classes one by one (specifying class names as config location) as well
 * as for classpath scanning (specifying base packages as config location).

Unfortunately that isn't going to work with just plain XML configuration.

Spring Kafka doesn't provide any hooks for XML definitions.

Upvotes: 1

Related Questions