Support kafka parallel-consumer

[bojanv55]

Expected Behavior

Enable parallel processing of the messages in single consumer.

Current Behavior

Single-threaded processing of the messages.

Context

Are there any plans to include ParallelStreamProcessor as an option for spring-kafka? https://github.com/confluentinc/parallel-consumer It should handle automatically multiple threads, acking and other stuff (synchronization on same key if needed etc.).

Currently I can configure this manually, but would be easier I guess If I would only need to implement void poll(Consumer<PollContext<K, V>> usersVoidConsumptionFunction);, and the rest is taken care by the spring.

[garyrussell]

No plans currently; but contributions are welcome!

[garyrussell]

That said; it's easy enough to use it. from Spring - just set up a @KafkaListener as normal; set the auto startup to "false" and then invoke the listener from the parallel consumer:

@SpringBootApplication
public class Kgh2381Application {

	private static final Logger log = LoggerFactory.getLogger(Kgh2381Application.class);

	public static void main(String[] args) {
		SpringApplication.run(Kgh2381Application.class, args);
	}

	@KafkaListener(id = "kgh2381", topics = "kgh2381", autoStartup = "false")
	void listen(String in) {
		log.info(in);
	}

	@Bean
	public NewTopic topic() {
		return TopicBuilder.name("kgh2381").partitions(1).replicas(1).build();
	}

	@Bean
	ApplicationRunner runner(KafkaListenerEndpointRegistry registry, ConsumerFactory<String, String> cf,
			KafkaTemplate<String, String> template) {

		return args -> {
			MessageListener messageListener = (MessageListener) registry.getListenerContainer("kgh2381")
					.getContainerProperties().getMessageListener();
			Consumer<String, String> consumer = cf.createConsumer("group", "");
			var options = ParallelConsumerOptions.<String, String>builder()
					.ordering(ProcessingOrder.KEY)
					.consumer(consumer)
					.maxConcurrency(10)
					.build();
			ParallelStreamProcessor<String, String> processor = ParallelStreamProcessor
					.createEosStreamProcessor(options);
			processor.subscribe(List.of("kgh2381"));
			processor.poll(context -> messageListener.onMessage(context.getSingleConsumerRecord(), null, consumer));
			IntStream.range(0, 10).forEach(i -> template.send("kgh2381", "foo" + i));
		};
	}

}

[dixitsingla]

I am trying to implement the parallel consumer in spring boot and using the above mentioned way. I have main class in separate java file and rest of the methods(KafkaListener, ApplicationRunner) in another java file.

Also, have created a ConsumerProp class with the below content:

@EnableKafka
@Configuration
public class ConsumerProp {

    @Bean
    public ConsumerFactory<String, String> consumerFactory() {

        Map<String, Object> configs = new HashMap<>();
        configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka:9092");
        configs.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configs.put(ConsumerConfig.GROUP_ID_CONFIG, "group_1");

        return new DefaultKafkaConsumerFactory<>(configs);
    }
}

But the spring boot just exists without even any error. Could you please help or put a sample working example which I can leverage for further use.

[garyrussell]

I assume you mean exits (not exists).

You must have done something wrong; the example above is a complete working example using Spring Boot. This is the output when running it...

  .   ____          _            __ _ _
 /\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
 \\/  ___)| |_)| | | | | || (_| |  ) ) ) )
  '  |____| .__|_| |_|_| |_\__, | / / / /
 =========|_|==============|___/=/_/_/_/
�[32m :: Spring Boot :: �[39m              �[2m (v2.7.3)�[0;39m

2023-06-29 14:28:45,673 [main] Starting Kgh2381Application using Java 11.0.11 on grussell3MD6R.vmware.com with PID 83586 (/Users/grussell/Development/stsws43/kgh2381/target/classes started by grussell in /Users/grussell/Development/stsws43/kgh2381)
2023-06-29 14:28:45,675 [main] No active profile set, falling back to 1 default profile: "default"
2023-06-29 14:28:46,176 [main] AdminClientConfig values: 
...
2023-06-29 14:28:51,533 [pc-broker-poll] pl.tlinkowski.unij.service.api.collect.UnmodifiableMapFactory service: selected pl.tlinkowski.unij.service.collect.jdk8.Jdk8UnmodifiableMapFactory (priority=40)
2023-06-29 14:28:53,520 [pc-broker-poll] [Consumer clientId=consumer-group-1, groupId=group] Successfully joined group with generation Generation{generationId=4, memberId='consumer-group-1-8ebc709a-c397-4585-89d2-51fc00808899', protocol='range'}
2023-06-29 14:28:53,522 [pc-broker-poll] [Consumer clientId=consumer-group-1, groupId=group] Finished assignment for group at generation 4: {consumer-group-1-8ebc709a-c397-4585-89d2-51fc00808899=Assignment(partitions=[kgh2381-0])}
2023-06-29 14:28:53,527 [pc-broker-poll] [Consumer clientId=consumer-group-1, groupId=group] Successfully synced group in generation Generation{generationId=4, memberId='consumer-group-1-8ebc709a-c397-4585-89d2-51fc00808899', protocol='range'}
2023-06-29 14:28:53,527 [pc-broker-poll] [Consumer clientId=consumer-group-1, groupId=group] Notifying assignor about the new Assignment(partitions=[kgh2381-0])
2023-06-29 14:28:53,529 [pc-broker-poll] [Consumer clientId=consumer-group-1, groupId=group] Adding newly assigned partitions: kgh2381-0
2023-06-29 14:28:53,529 [pc-broker-poll] Assigned 1 total (1 new) partition(s) [kgh2381-0]
2023-06-29 14:28:53,540 [pc-broker-poll] [Consumer clientId=consumer-group-1, groupId=group] Setting offset for partition kgh2381-0 to the committed offset FetchPosition{offset=30, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[localhost:9092 (id: 0 rack: null)], epoch=0}}
2023-06-29 14:28:53,559 [pc-pool-1-thread-1] foo0
2023-06-29 14:28:53,560 [pc-pool-1-thread-1] pl.tlinkowski.unij.service.api.collect.UnmodifiableListFactory service: selected pl.tlinkowski.unij.service.collect.jdk8.Jdk8UnmodifiableListFactory (priority=40)
2023-06-29 14:28:53,562 [pc-pool-1-thread-2] foo1
2023-06-29 14:28:53,564 [pc-pool-1-thread-3] foo2
2023-06-29 14:28:53,566 [pc-pool-1-thread-4] foo3
2023-06-29 14:28:53,567 [pc-pool-1-thread-5] foo4
2023-06-29 14:28:53,569 [pc-pool-1-thread-6] foo5
2023-06-29 14:28:53,571 [pc-pool-1-thread-7] foo6
2023-06-29 14:28:53,573 [pc-pool-1-thread-8] foo7
2023-06-29 14:28:53,574 [pc-pool-1-thread-9] foo8
2023-06-29 14:28:53,576 [pc-pool-1-thread-10] foo9

[garyrussell]

This is its application.properties

spring.kafka.consumer.auto-offset-reset=earliest
spring.kafka.consumer.enable-auto-commit=false

... rest of the methods(KafkaListener, ApplicationRunner) in another java file.

That file must be annotated with @Configuration so that the beans are registered.

You don't need your own consumer factory bean, just use Boot's.

[dixitsingla]

Thank for the correcting the type(exits).

I was able to run the application by specifying @Autowired annotation for the ApplicationRunner runner( method. Although I have tried @Configuration annotation as well and it's also working.

I am using consumer factory bean as I need to provide few properties dynamically(some cases from environment variables). Like Bootstrap server etc..

If that's okay with you I really would like you to share the sample project you just ran above.

Thanks again for all the help.

[garyrussell]

If that's okay with you I really would like you to share the sample project you just ran above.

There's nothing more to share; that is the entire app (together with the properties above).

[dixitsingla]

As you stated earlier 'You don't need your own consumer factory bean, just use Boot's.'

I have just observed that the runner method is using the spring boot's consumer factory only, the one I have created is not being used. Could you please help me what should I do use the consumer factory I have created.

[garyrussell]

Since you define your own consumer factory bean, Boot will detect it and not declare its own; yours will be injected into the runner instead.

[garyrussell]

KafkaAutoConfiguration:

@Bean
@ConditionalOnMissingBean(ConsumerFactory.class)
public DefaultKafkaConsumerFactory<?, ?> kafkaConsumerFactory(

Your @Configuration class must be in the same (or child) package as the @SpringBootApplication.

[dixitsingla]

As of now I am trying with the spring boot's consumer factory.

My application is stuck at
2023-07-03 22:19:38.423 INFO 40836 --- [ pc-broker-poll] o.a.k.c.c.internals.SubscriptionState : [Consumer clientId=consumer-group-1, groupId=group] Resetting offset for partition topic_1-0 to position FetchPosition{offset=6, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[127.0.0.1:9092 (id: 1 rack: null)], epoch=0}}.
I don't see any consumed messages. there are 6 messages in the topic. Seems like the listen method is not called at all.

@KafkaListener(id = "group_1", topics = "topic_1", autoStartup = "false")
    void listen(String record) {
        log.info(record);
    }

I am not very good at spring boot yet, Sorry for asking the small doubts.