Kafka Listener Method is not invoked. Consumer not consuming.

Question

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();
}

}

Artem Bilan Artem Bilan · Accepted Answer · 2017-10-04T20:35:38

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.

Kafka Listener Method is not invoked. Consumer not consuming.

1 Answers