Parallel consumer with spring boot

[amber13574]

Hi Folks,

I'm currently working on my service where I am testing how the parallel-consumer library (https://github.com/confluentinc/parallel-consumer) behaves with spring. However, I've encountered an issue while setting up the factory container customizer. I want to set my message listener, which is of type ParallelStreamProcessor. However, the method setupMessageListener only expects AcknowledgingMessageListener or MessageListener. Is there any way to address this? One option is to create a customized listener by extending one of the existing listeners, but I'm open to other solutions.

/**
     * Set the message listener; must be a {@link org.springframework.kafka.listener.MessageListener}
     * or {@link org.springframework.kafka.listener.AcknowledgingMessageListener}.
     * @param messageListener the listener.
     */
    public void setMessageListener(Object messageListener) {
        this.messageListener = messageListener;
        adviseListenerIfNeeded();
    }

This is how I am creating beans

@Bean
  public ParallelStreamProcessor<String, String> parallelStreamProcessor(Consumer<String, String> kafkaConsumer) {

    ParallelConsumerOptions<String, String> options = ParallelConsumerOptions.<String, String>builder()
        .ordering(ParallelConsumerOptions.ProcessingOrder.KEY)
        .maxConcurrency(16)
        .commitMode(ParallelConsumerOptions.CommitMode.PERIODIC_CONSUMER_SYNC)
        .consumer(kafkaConsumer)
        .build();

    return ParallelStreamProcessor.createEosStreamProcessor(options);
  }

  @Bean
  public ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerContainerFactory(
      ConsumerFactory<String, String> consumerFactory,
      ParallelStreamProcessor<String, String> parallelStreamProcessor) {
    ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
    factory.setConsumerFactory(consumerFactory);

    factory.setContainerCustomizer(container -> {
      container.setupMessageListener(parallelStreamProcessor);
      container.setAutoStartup(true);
    });

    return factory;
  }

I expect alternate solution to resolve this issue.

As I can see there are no example present at the moment , but feel free to add example If anyone has implemented the confluent parallel-consumer with spring.

[rkolesnev]

Hi @amber13574,
Honestly i haven't had integrated Parallel Consumer with Spring through Spring factory / customizers properly before.
Have you figured out a good way to do it?
I would really appreciate a full example to add it to docs and integration examples.