Spring Cloud Stream Binder

Overview

The kubemq-spring-cloud-stream-binder module lets you drive KubeMQ from Spring Cloud Stream's functional programming model. Instead of injecting KubeMQTemplate or wiring @KubeMQ*Listener methods, you declare plain Supplier, Function, and Consumer beans and let the framework bind their inputs and outputs to KubeMQ channels.

The binder registers itself with Spring Cloud Stream as kubemq through a META-INF/spring.binders descriptor:

kubemq:\
io.kubemq.spring.cloud.stream.binder.config.KubeMQBinderConfiguration

When the binder is on the classpath, Spring Cloud Stream discovers it under the name kubemq. You then map function inputs and outputs to KubeMQ destinations, and choose — per binding — whether each destination is an Events, Events Store, or Queues channel. For the semantics of each pattern, see the core Events, Events Store, and Queues concepts.

The following diagram shows a single Function bean wired between two KubeMQ channels. The binder subscribes the input channel, hands each message to the function, and publishes the result to the output channel.

A function bean is bound between two KubeMQ channels; the binder consumes, invokes the function, and publishes the result.

Prerequisites

You need a running KubeMQ broker. The binder connects over the native gRPC port 50000, the same port the Spring Boot starter uses. The shared HTTP/REST and dashboard endpoints listen on 9090:

docker run -d \  --name kubemq \  -p 50000:50000 \  -p 9090:9090 \  -e KUBEMQ_TOKEN=YOUR_LICENSE_KEY \  europe-docker.pkg.dev/kubemq/images/kubemq:next

This guide targets Java 17+, Spring Boot 3.2.0+, and the Spring Cloud 2023.0.0 release train.

Add the Dependency

Add the binder alongside the KubeMQ starter and Spring Cloud Stream, and import the Spring Cloud 2023.0.0 BOM so the Spring Cloud Stream version is managed for you. The binder is published under the io.kubemq group, version 1.0.0.

dependencyManagement {
    imports {
        mavenBom("org.springframework.cloud:spring-cloud-dependencies:2023.0.0")
    }
}

dependencies {
    implementation("io.kubemq:kubemq-spring-boot-starter:1.0.0")
    implementation("io.kubemq:kubemq-spring-cloud-stream-binder:1.0.0")
    implementation("org.springframework.cloud:spring-cloud-stream")
}

<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>org.springframework.cloud</groupId>
      <artifactId>spring-cloud-dependencies</artifactId>
      <version>2023.0.0</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>

<dependencies>
  <dependency>
    <groupId>io.kubemq</groupId>
    <artifactId>kubemq-spring-boot-starter</artifactId>
    <version>1.0.0</version>
  </dependency>
  <dependency>
    <groupId>io.kubemq</groupId>
    <artifactId>kubemq-spring-cloud-stream-binder</artifactId>
    <version>1.0.0</version>
  </dependency>
  <dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
  </dependency>
</dependencies>

Functional Binding Configuration

Spring Cloud Stream binds the inputs and outputs of your function beans by convention. For a Function<I, O> bean named uppercase, the input binding is uppercase-in-0 and the output binding is uppercase-out-0. You map each binding to a KubeMQ channel with destination, and optionally a consumer group with group.

Set spring.cloud.stream.default-binder: kubemq so every binding routes through the KubeMQ binder, and declare the active functions with spring.cloud.function.definition.

spring:
  application:
    name: kubemq-example-spring-cloud-stream-events
  main:
    web-application-type: none
  cloud:
    function:
      definition: uppercase
    stream:
      default-binder: kubemq
      bindings:
        uppercase-in-0:
          destination: spring-scs.events.in
          group: scs-events-demo
        uppercase-out-0:
          destination: spring-scs.events.out
      kubemq:
        default:
          consumer:
            pattern: EVENTS
          producer:
            pattern: EVENTS

kubemq:
  address: ${KUBEMQ_ADDRESS:localhost:50000}
  client-id: spring-scs-events

Here uppercase-in-0 consumes from the spring-scs.events.in channel (load-balanced within the scs-events-demo group), the function transforms the payload, and uppercase-out-0 publishes the result to spring-scs.events.out. The kubemq.* block at the bottom is the standard starter connection configuration — the binder reuses it.

Pattern Selection

KubeMQ exposes several messaging patterns on the same broker, so the binder needs to know which pattern backs each destination. You select it per binding with the binder-specific pattern property, which is one of the KubeMQPattern enum values:

public enum KubeMQPattern {
    EVENTS,
    EVENTS_STORE,
    QUEUES
}

Set the pattern independently for the consumer and the producer side of a binding:

spring:
  cloud:
    stream:
      kubemq:
        bindings:
          uppercase-in-0:
            consumer:
              pattern: EVENTS
          uppercase-out-0:
            producer:
              pattern: EVENTS

To apply the same pattern to every binding without repeating it, use the default block (as in the example above):

spring:
  cloud:
    stream:
      kubemq:
        default:
          consumer:
            pattern: EVENTS_STORE
          producer:
            pattern: EVENTS_STORE

Both consumer.pattern and producer.pattern default to EVENTS when omitted.

Consumer Properties

The consumer side accepts additional KubeMQ-specific properties under spring.cloud.stream.kubemq.bindings.<binding>.consumer.* (or under default.consumer.*). These map to the KubeMQConsumerProperties class.

The eventsStoreType and eventsStoreSequenceValue properties apply only to EVENTS_STORE bindings; the poll*, visibilitySeconds, and autoAckMessages properties apply only to QUEUES bindings.

spring:
  cloud:
    stream:
      kubemq:
        bindings:
          process-in-0:
            consumer:
              pattern: QUEUES
              pollMaxMessages: 10
              pollWaitTimeoutInSeconds: 5
              visibilitySeconds: 30
              autoAckMessages: false

How the Binder Routes a Pattern

Each pattern is backed by its own message-driven adapter in the binder's adapter package. The adapter subscribes (or polls) the KubeMQ channel for a consumer binding and forwards every received message into the Spring Cloud Stream input channel that feeds your function.

The pattern you set on a binding decides which of these three adapters the binder instantiates. You do not interact with the adapters directly — they are the binder's internal bridge between KubeMQ subscriptions and Spring's message channels.

Header Mapping

Spring messages carry headers; KubeMQ messages carry string tags. The binder's KubeMQHeaderMapper translates between the two in both directions.

• Outbound (Spring to KubeMQ): each Spring header becomes a KubeMQ tag. The framework headers id, timestamp, and contentType are prefixed with spring- to avoid collisions, and trace-propagation headers (kubemq-traceparent, kubemq-tracestate) are passed through unchanged.
• Inbound (KubeMQ to Spring): each tag becomes a Spring header. Tags that were prefixed with spring- are restored to their original header names. Tags already carrying a recognized prefix (kubemq-, spring-, app-) are kept as-is; any other tag is namespaced under kubemq.tag.<name> so application metadata stays distinct from framework headers.

This means a header you add on the producing side — for example app-tenant-id — survives the round trip and is readable as a header on the consuming side.

Channel Provisioning

KubeMQ channels are created automatically on first use, so the binder does not need to pre-create anything. Its KubeMQChannelProvisioner is effectively a no-op that wraps the configured channel name in a KubeMQDestination for both producer and consumer destinations:

public class KubeMQDestination implements ProducerDestination, ConsumerDestination {

    private final String name;

    public KubeMQDestination(String name) {
        this.name = name;
    }

    @Override
    public String getName() {
        return name;
    }
}

There is nothing to declare ahead of time — set a destination on your binding and the channel materializes the first time a message flows.

Worked Example: Events

The spring-cloud-stream-events example wires an uppercase Function<String, String> from an Events input channel to an Events output channel. It uses the configuration shown earlier in Functional Binding Configuration.

package io.kubemq.spring.boot.examples.scs;

import java.util.function.Function;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class ScsEventsFunctionConfig {

    private static final Logger log = LoggerFactory.getLogger(ScsEventsFunctionConfig.class);

    @Bean
    public Function<String, String> uppercase() {
        return payload -> {
            if (payload == null) {
                return null;
            }
            String result = payload.toUpperCase();
            log.info("SCS function: {} -> {}", payload, result);
            return result;
        };
    }
}

package io.kubemq.spring.boot.examples.scs;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.stereotype.Component;

@Component
public class ScsEventsRunner implements ApplicationRunner {

    private static final Logger log = LoggerFactory.getLogger(ScsEventsRunner.class);

    private final StreamBridge streamBridge;

    public ScsEventsRunner(StreamBridge streamBridge) {
        this.streamBridge = streamBridge;
    }

    @Override
    public void run(ApplicationArguments args) throws InterruptedException {
        boolean sent = streamBridge.send("uppercase-in-0", "hello from spring cloud stream");
        log.info("StreamBridge send result: {}", sent);
        Thread.sleep(1000);
        log.info("Spring Cloud Stream events example completed.");
    }
}

INFO  StreamBridge send result: true
INFO  SCS function: hello from spring cloud stream -> HELLO FROM SPRING CLOUD STREAM
INFO  Spring Cloud Stream events example completed.

Queues and Events Store Variants

Two sibling examples reuse the exact same shape — a single Function bean wired between an input and an output binding — and differ only in the function logic and the pattern they select.

spring:
  cloud:
    function:
      definition: process
    stream:
      default-binder: kubemq
      bindings:
        process-in-0:
          destination: spring-scs.queues.in
          group: scs-queues-demo
        process-out-0:
          destination: spring-scs.queues.out
      kubemq:
        default:
          consumer:
            pattern: QUEUES
          producer:
            pattern: QUEUES

kubemq:
  address: ${KUBEMQ_ADDRESS:localhost:50000}
  client-id: spring-scs-queues

@Bean
public Function<String, String> process() {
    return payload -> {
        if (payload == null) {
            return null;
        }
        String result = "processed: " + payload;
        log.info("SCS queues function: {} -> {}", payload, result);
        return result;
    };
}

spring:
  cloud:
    function:
      definition: transform
    stream:
      default-binder: kubemq
      bindings:
        transform-in-0:
          destination: spring-scs.events-store.in
          group: scs-events-store-demo
        transform-out-0:
          destination: spring-scs.events-store.out
      kubemq:
        default:
          consumer:
            pattern: EVENTS_STORE
          producer:
            pattern: EVENTS_STORE

kubemq:
  address: ${KUBEMQ_ADDRESS:localhost:50000}
  client-id: spring-scs-events-store

@Bean
public Function<String, String> transform() {
    return payload -> {
        String result = payload.replace(" ", "-");
        log.info("SCS events-store function: {} -> {}", payload, result);
        return result;
    };
}

When to Use the Binder

The binder and the imperative KubeMQTemplate / @KubeMQ*Listener API both target the same broker — choose based on how your application is structured.

Related