spring-cloud / spring-cloud/spring-cloud-stream
Expose KafkaConsumer in ReceiverOptionsCustomizer.addAssignListener() - parity with KafkaBindingRebalanceListener
@garyrussell is already working on this.
Since Jul 20, 2023.
- 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
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Assessment
This issue has not been assessed yet.