From f48523f774b814a187d02855a13f9bb54952bf18 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 5 Apr 2023 15:50:07 -0400 Subject: [PATCH] GH-2691: Kafka Reactive Binder Support Multiplex (#2698) * GH-2691: Kafka Reactive Binder Support Multiplex Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2691 * Add docs. --- .../binder/reactorkafka/ReactorKafkaBinder.java | 14 +++++++++++--- .../reactorkafka/ReactorKafkaBinderTests.java | 12 +++++++----- docs/src/main/asciidoc/kafka/kafka-reactive.adoc | 5 +++++ 3 files changed, 23 insertions(+), 8 deletions(-) diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java index 5da1e0b55..b4ed9ef84 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/main/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinder.java @@ -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 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 destList = Arrays.stream(StringUtils.commaDelimitedListToStringArray(destinations)) + .map(dest -> dest.trim()) + .toList(); ReceiverOptions 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 finalOpts = opts; diff --git a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java index 4d295ab99..cdc2cdf19 100644 --- a/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java +++ b/binders/kafka-binder/spring-cloud-stream-binder-kafka-reactive/src/test/java/org/springframework/cloud/stream/binder/reactorkafka/ReactorKafkaBinderTests.java @@ -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> sub = new Subscriber>() { @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 props = new ExtendedConsumerProperties(ext); + props.setMultiplex(true); - Binding consumer = binder.bindConsumer("testC", "foo", inbound, props); + Binding 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(); diff --git a/docs/src/main/asciidoc/kafka/kafka-reactive.adoc b/docs/src/main/asciidoc/kafka/kafka-reactive.adoc index ec2fe951f..634e9ce6e 100644 --- a/docs/src/main/asciidoc/kafka/kafka-reactive.adoc +++ b/docs/src/main/asciidoc/kafka/kafka-reactive.adoc @@ -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.