CODEWITHSUNDEEP
springspring-kafka
Mike Snare
Mike Snare

Reputation: 423

Spring Kafka listenerExecutor

I'm setting up a kafka listener in a spring boot application and I can't seem to get the listener running in a pool using an executor. Here's my kafka configuration:

@Bean
ThreadPoolTaskExecutor messageProcessorExecutor() {
    logger.info("Creating a message processor pool with {} threads", numThreads);
    ThreadPoolTaskExecutor exec = new ThreadPoolTaskExecutor();
    exec.setCorePoolSize(200);
    exec.setMaxPoolSize(200);
    exec.setKeepAliveSeconds(30);
    exec.setAllowCoreThreadTimeOut(true);
    exec.setQueueCapacity(0); // Yields a SynchronousQueue
    exec.setThreadFactory(ThreadFactoryFactory.defaultNamingFactory("kafka", "processor"));
    return exec;
}

@Bean
public ConsumerFactory<String, PollerJob> consumerFactory() {
    Map<String, Object> props = new HashMap<>();
    props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
    props.put(ConsumerConfig.GROUP_ID_CONFIG, consumerGroup);
    props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
    DefaultKafkaConsumerFactory<String, PollerJob> factory = new DefaultKafkaConsumerFactory<>(props,
            new StringDeserializer(),
            new JsonDeserializer<>(PollerJob.class));
    return factory;
}

@Bean
public ConcurrentKafkaListenerContainerFactory<String, PollerJob> kafkaListenerContainerFactory() {
    ConcurrentKafkaListenerContainerFactory<String, PollerJob> factory
            = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory());
    factory.setConcurrency(Integer.valueOf(kafkaThreads));
    factory.getContainerProperties().setListenerTaskExecutor(messageProcessorExecutor());
    factory.getContainerProperties().setAckMode(AbstractMessageListenerContainer.AckMode.MANUAL);
    return factory;
}

The ThreadFactoryFactory used by the ThreadPoolTaskExecutor just makes sure the thread is named like 'kafka-1-processor-1'.

The ConsumerFactory has the ENABLE_AUTO_COMMIT_CONFIG flag set to false and I'm using manual mode for the acknowledgement which is required to use executors according to the documentation.

My listener looks like this:

@KafkaListener(topics = "my_topic",
        group = "my_group",
        containerFactory = "kafkaListenerContainerFactory")
public void listen(@Payload SomeJob job, Acknowledgment ack) {
    ack.acknowledge();
    logger.info("Running job {}", job.getId());
    ....
}

Using the Admin Server I can inspect all the threads and only one kafka-N-processor-N threads is being created but I expected to see up to 200. The jobs are all running one at a time on the that one thread and I can't figure out why.

How can I get this setup to run the listeners using my executor with as many threads as possible?

I'm using Spring Boot 1.5.4.RELEASE and kafka 0.11.0.0.

Upvotes: 3

Views: 4400

Answers (1)

Artem Bilan
Artem Bilan

Reputation: 121272

If your topic has only one partition, according the consumer group policy, only one consumer is able to poll that partition.

The ConcurrentMessageListenerContainer indeed creates as much target KafkaMessageListenerContainer instances as provided concurrency. And it does that only in case it doesn't know the number of partitions in the topic.

When the rebalance in consumer group happens only one consumer gets partition for consuming. All the work is really done there in a single thread:

 private void startInvoker() {
    ListenerConsumer.this.invoker = new ListenerInvoker();
    ListenerConsumer.this.listenerInvokerFuture = this.containerProperties.getListenerTaskExecutor()
            .submit(ListenerConsumer.this.invoker);
}

One partition - one thread for sequential records processing.

Upvotes: 1

Related Questions