spring-cloud / spring-cloud/spring-cloud-stream

Expose KafkaConsumer in ReceiverOptionsCustomizer.addAssignListener() - parity with KafkaBindingRebalanceListener

Open
#2,727 6 comments 7 reactions 1 assignee View on GitHub

@garyrussell is already working on this.

Since Jul 20, 2023.

documentation REACTIVE-KAFKA-BINDER
Dominant language
Java
Stars
1.1k
Forks
646
Avg merge
2d 3h
Merged PRs (30d)
8

Description

Feature Request:

To be able to access the KafkaConsumer when using receiverOptions.addAssignListener:

  @Bean
  public ReceiverOptionsCustomizer customizer() {
    return (bindingName, receiverOptions) -> receiverOptions.addAssignListener(receiverPartitions -> {
      receiverPartitions.forEach(part -> {
      ...

Under the hood this does actually return an instance of SeekablePartition which contains the consumer - unfortunately it isn't publicly accessible.

Why is this useful?

It would provide parity with KafkaBindingRebalanceListener which exposes the hook below.

  @Override
  public void onPartitionsAssigned(String bindingName, Consumer<?, ?> consumer, Collection<TopicPartition> partitions, boolean initial) {

This is useful for microservices that hydrate a cache from Kafka on start-up - as we can ultimately tie consumer.endOffsets(partitions) to a readiness probe once the last offset (per partition) has been consumed.

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.