DEV Community

Roger Viñas Alcon
Roger Viñas Alcon

Posted on • Updated on


Spring Cloud Stream Multibinder

Spring Cloud Stream is the solution provided by Spring to build applications connected to shared messaging systems.

It offers an abstraction (the binding) that works the same whatever underneath implementation we use (the binder):

  • Apache Kafka
  • Rabbit MQ
  • Kafka Streams
  • Amazon Kinesis
  • ...

But what if we need more than one binder in the same application? 🤔

Not a problem! You can specify multiple binder configurations as documented in Connecting to Multiple Systems

Let's put the theory into practice 🛠️ ...

GitHub logo rogervinas / spring-cloud-stream-multibinder

🍀 Spring Cloud Stream Multibinder - Kafka & Kafka Streams


We want to implement this flow:

Alt Text

  • User will POST string payloads to /text endpoint
  • A KafkaProducer will send these payloads to topic pub.texts as { "text" : string }
  • A KafkaStreams transformation will consume from topic pub.texts and produce events to topic pub.lengths as { "length" : number }
  • A KafkaConsumer will consume events from topic pub.lengths and log them to the console

So we are using two Spring Cloud Stream binders:

  • Kafka
  • Kafka Streams

We create a project using this spring initializr configuration and we add:

  • Kafka binder lib spring-cloud-stream-binder-kafka
  • Kafka Streams binder lib spring-cloud-stream-binder-kafka-streams

Integration Test

We start writing the following integration test using:

fun `should process text lengths`(capturedOutput: CapturedOutput) {
  postText("Or do not")
  postText("There is no try")

  await().atMost(ONE_MINUTE).untilAsserted {
    assertThat(capturedOutput.out).contains("Consumed length [2]")
    assertThat(capturedOutput.out).contains("Consumed length [9]")
    assertThat(capturedOutput.out).contains("Consumed length [15]")
Enter fullscreen mode Exit fullscreen mode

This test will obviously fail, but it should work once we have finished our implementation.

Spring Cloud Stream binders configuration

Next we configure the two binders we are using:

    name: "spring-cloud-stream-multibinder"
        definition: textProducer;textLengthProcessor;lengthConsumer
          destination: "${kafka.topic.texts}"
          binder: kafka1
          destination: "${kafka.topic.texts}"
          binder: kstream1
          destination: "${kafka.topic.lengths}"
          binder: kstream1
          destination: "${kafka.topic.lengths}"
          group: "${}"
          binder: kafka1
          type: kafka
              brokers: "${kafka.brokers}"
          type: kstream
              applicationId: "${}-KApp"
              brokers: "${kafka.brokers}"

    texts: "pub.texts"
    lengths: "pub.lengths"
  brokers: "localhost:9094"
Enter fullscreen mode Exit fullscreen mode

With this configuration:

  • Spring Cloud Stream will create:
    • A Kafka Streams binder connected to localhost:9094
    • A Kafka binder connected to localhost:9094
  • Following the Spring Cloud Stream functional programming model conventions we need to create:
    • An instance named textProducer that should implement Supplier<Flux<TextEvent>> interface
    • An instance named textLengthProcessor that should implement Function<KStream<String, TextEvent>, KStream<String, LengthEvent>> interface
    • An instance named lengthConsumer that should implement Consumer<LengthEvent> interface

💡 We use different values for the Kafka Streams applicationId and the Kafka Consumers group to avoid undesired behaviors.
💡 We are using Spring Cloud Stream's default serialization/deserialization of Kotlin data classes to Json. In order for this to work we need to add com.fasterxml.jackson.module:jackson-module-kotlin dependency.
💡 You can find all the available configuration properties documented in:


First we need this TextProducer interface that we will implement later:

data class TextEvent(val text: String)

interface TextProducer {

  fun produce(event: TextEvent)
Enter fullscreen mode Exit fullscreen mode


Once we have the test ...

@WebFluxTest(controllers = [TextController::class])
class TextControllerTest {

  lateinit var webClient: WebTestClient

  lateinit var textProducer: TextProducer

  fun `should produce text events`() {
    val text = "Some awesome text"

Enter fullscreen mode Exit fullscreen mode

... the implementation is easy:

class TextController(private val textProducer: TextProducer) {

  @PostMapping("/text", consumes = [TEXT_PLAIN_VALUE])
  fun text(@RequestBody text: String) {
Enter fullscreen mode Exit fullscreen mode

TextProducer implementation

We implement TextProducer using Spring Cloud Stream like this:

class TextStreamProducer : Supplier<Flux<TextEvent>>, TextProducer {

  private val sink = Sinks.many().unicast()

  override fun produce(event: TextEvent) {
    sink.emitNext(event, FAIL_FAST)

  override fun get() = sink.asFlux()
Enter fullscreen mode Exit fullscreen mode

And we can easily test the implementation as follows:

fun `should produce text events`() {
  val producer = TextStreamProducer()

  val events = mutableListOf<TextEvent>()

  producer.produce(TextEvent("nobody is"))

    TextEvent("nobody is"),
Enter fullscreen mode Exit fullscreen mode


We implement the transformation using Kafka Stream's mapValues method:

class TextLengthProcessor : Function<KStream<String, TextEvent>, KStream<String, LengthEvent>> {

  override fun apply(input: KStream<String, TextEvent>): KStream<String, LengthEvent> {
    return input
      .mapValues { event -> LengthEvent(event.text.length) }
Enter fullscreen mode Exit fullscreen mode

And we can test it using kafka-streams-test-utils:

private const val TOPIC_IN = ""
private const val TOPIC_OUT = "topic.out"

private const val KEY1 = "key1"
private const val KEY2 = "key2"
private const val KEY3 = "key3"

internal class TextLengthProcessorTest {

  private lateinit var topologyTestDriver: TopologyTestDriver
  private lateinit var topicIn: TestInputTopic<String, TextEvent>
  private lateinit var topicOut: TestOutputTopic<String, LengthEvent>

  fun beforeEach() {
    val stringSerde = Serdes.StringSerde()
    val streamsBuilder = StreamsBuilder()


    val config = Properties().apply {
      setProperty(StreamsConfig.APPLICATION_ID_CONFIG, "test")
      setProperty(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "test-server")
      setProperty(JsonDeserializer.TRUSTED_PACKAGES, "*")
    val topology =
    topologyTestDriver = TopologyTestDriver(topology, config)
    topicIn = topologyTestDriver.createInputTopic(TOPIC_IN, stringSerde.serializer(), JsonSerde(
    topicOut = topologyTestDriver.createOutputTopic(TOPIC_OUT, stringSerde.deserializer(), JsonSerde(

  fun afterEach() {

  fun `should produce length events from text events`() {
    topicIn.pipeInput(KEY1, TextEvent("Hello!"))
    topicIn.pipeInput(KEY2, TextEvent("How are you?"))
    topicIn.pipeInput(KEY3, TextEvent("Bye!"))

      KeyValue(KEY1, LengthEvent(6)),
      KeyValue(KEY2, LengthEvent(12)),
      KeyValue(KEY3, LengthEvent(4))
Enter fullscreen mode Exit fullscreen mode


LengthStreamConsumer implements Consumer as expected by Spring Cloud Stream conventions. We decouple the final implementation using the interface LengthProcessor:

data class LengthEvent(val length: Int)

interface LengthProcessor {

  fun process(event: LengthEvent)

class LengthStreamConsumer(private val processor: LengthProcessor) : Consumer<LengthEvent> {

  override fun accept(event: LengthEvent) {
Enter fullscreen mode Exit fullscreen mode

And we can test it with this code:

fun `should consume length events`() {
  val lengthProcessor = mock(
  val lengthStreamConsumer = LengthStreamConsumer(lengthProcessor)


Enter fullscreen mode Exit fullscreen mode

LengthConsumer implementation

For this demo the implementation just logs the event:

class LengthConsoleProcessor : LengthProcessor {

  private val logger = LoggerFactory.getLogger(

  override fun process(event: LengthEvent) {"Consumed length [${event.length}]")
Enter fullscreen mode Exit fullscreen mode

And we can also test it using JUnit Jupiter OutputCaptureExtension:

internal class LengthConsoleProcessorTest {

  fun `should log consumed length event to console`(capturedOutput: CapturedOutput) {
    val lengthConsoleProcessor = LengthConsoleProcessor()


    assertThat(capturedOutput.out).contains("Consumed length [53]")
Enter fullscreen mode Exit fullscreen mode

Wiring it all together

We only need to create all the required instances naming them accordingly to the binder configuration:

class MyApplicationConfiguration {

  fun textProducer() = TextStreamProducer()

  fun textLengthProcessor(): Function<KStream<String, TextEvent>, KStream<String, LengthEvent>> = TextLengthProcessor()

  fun lengthConsumer(lengthProcessor: LengthProcessor) = LengthStreamConsumer(lengthProcessor)

  fun lengthProcessor() = LengthConsoleProcessor()
Enter fullscreen mode Exit fullscreen mode

🤷 textLengthProcessor type should be specified explicitly, if not the Kafka Streams binder will not use it.

And that is it, now MyApplicationIntegrationTest should work! 🤞

Happy coding!

Top comments (0)

Timeless DEV post...

Git Concepts I Wish I Knew Years Ago

The most used technology by developers is not Javascript.

It's not Python or HTML.

It hardly even gets mentioned in interviews or listed as a pre-requisite for jobs.

I'm talking about Git and version control of course.

One does not simply learn git