SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2691: Kafka Reactive Binder Support Multiplex (#2698)

Browse Source 
* GH-2691: Kafka Reactive Binder Support Multiplex

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2691

* Add docs.
This commit is contained in:
Gary Russell
2023-04-05 15:50:07 -04:00
committed by GitHub
parent 92279db2a1
commit f48523f774
3 changed files with 23 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java
View File

@@ -17,7 +17,7 @@
package org.springframework.cloud.stream.binder.reactorkafka;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.UUID;
@@ -181,16 +181,24 @@ public class ReactorKafkaBinder
Map<String, Object> configs = BindingUtils.createConsumerConfigs(anonymous, consumerGroup, properties,
this.configurationProperties);
String destinations = destination.getName();
if (this.consumerConfigCustomizer != null) {
this.consumerConfigCustomizer.configure(configs, properties.getBindingName(), destination.getName());
this.consumerConfigCustomizer.configure(configs, properties.getBindingName(), destinations);
}
MessageConverter converter = BindingUtils.getConsumerMessageConverter(getApplicationContext(), properties,
this.configurationProperties);
Assert.isInstanceOf(RecordMessageConverter.class, converter);
/*
* No need to check multiplex here because, if false, the topics are bound one-at-a-time;
* it is still required by the provisioner, however.
*/
List<String> destList = Arrays.stream(StringUtils.commaDelimitedListToStringArray(destinations))
.map(dest -> dest.trim())
.toList();
ReceiverOptions<Object, Object> opts = ReceiverOptions.create(configs)
.addAssignListener(parts -> logger.info("Assigned: " + parts))
.subscription(Collections.singletonList(destination.getName()));
.subscription(destList);
opts = this.receiverOptionsCustomizer.apply(properties.getBindingName(), opts);
ReceiverOptions<Object, Object> finalOpts = opts;

12
binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java
View File

@@ -66,7 +66,7 @@ import static org.mockito.Mockito.mock;
* @since 4.0
*
*/
@EmbeddedKafka(topics = { "testC", "testC1", "testP" })
@EmbeddedKafka(topics = { "testCa", "testCb", "testC1", "testP" })
public class ReactorKafkaBinderTests {
@SuppressWarnings({ "rawtypes", "unchecked" })
@@ -82,14 +82,14 @@ public class ReactorKafkaBinderTests {
ReactorKafkaBinder binder = new ReactorKafkaBinder(binderProps, provisioner);
binder.setApplicationContext(mock(GenericApplicationContext.class));
CountDownLatch latch = new CountDownLatch(1);
CountDownLatch latch = new CountDownLatch(2);
FluxMessageChannel inbound = new FluxMessageChannel();
Subscriber<Message<?>> sub = new Subscriber<Message<?>>() {
@Override
public void onSubscribe(Subscription s) {
s.request(1);
s.request(2);
}
@Override
@@ -111,13 +111,15 @@ public class ReactorKafkaBinderTests {
KafkaConsumerProperties ext = new KafkaConsumerProperties();
ExtendedConsumerProperties<KafkaConsumerProperties> props =
new ExtendedConsumerProperties<KafkaConsumerProperties>(ext);
props.setMultiplex(true);
Binding<MessageChannel> consumer = binder.bindConsumer("testC", "foo", inbound, props);
Binding<MessageChannel> consumer = binder.bindConsumer("testCa, testCb", "foo", inbound, props);
DefaultKafkaProducerFactory pf =
new DefaultKafkaProducerFactory<>(KafkaTestUtils.producerProps(EmbeddedKafkaCondition.getBroker()));
KafkaTemplate kt = new KafkaTemplate<>(pf);
kt.send("testC", "foo").get(10, TimeUnit.SECONDS);
kt.send("testCa", "foo").get(10, TimeUnit.SECONDS);
kt.send("testCb", "bar").get(10, TimeUnit.SECONDS);
assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue();
consumer.unbind();
pf.destroy();

5
docs/src/main/asciidoc/kafka/kafka-reactive.adoc
View File

@@ -172,3 +172,8 @@ spring.cloud.stream.bindings.lowercase-in-0.consumer.concurrency=3
```
That will create three dedicated `KafkaReceiver` objects that generate three separate `Flux` implementations and then stream them to the handler method.
=== Multiplex
Starting with version 4.0.3, the common consumer property `multiplex` is now supported by the reactive binder, where a single binding can consume from multiple topics.
When `false` (default), a separate binding is created for each topic specified in a comma-delimited list in the common `destination` property.