# Spring Cloud Stream Binder (/integrations/spring-boot/how-to/spring-cloud-stream-binder)



## Overview [#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:

```properties title="META-INF/spring.binders"
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](/learn/events), [Events Store](/learn/events-store), and [Queues](/learn/queues) concepts.

<Callout type="info">
  The binder reuses the same `PubSubClient` and `QueuesClient` beans created by `kubemq-spring-boot-starter`'s auto-configuration, so the binder and the rest of your Spring Boot app share one gRPC connection to the broker. Configure the connection once under `kubemq.*` as in any other Spring Boot KubeMQ app.
</Callout>

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.

<Mermaid
  chart="`
flowchart LR
  In[&#x22;KubeMQ channel<br/>(in)&#x22;] -->|consume| Adapter[&#x22;KubeMQ binder<br/>message-driven adapter&#x22;]
  Adapter --> Fn[&#x22;Function&lt;String, String&gt;<br/>bean&#x22;]
  Fn --> Producer[&#x22;KubeMQ binder<br/>producer&#x22;]
  Producer -->|publish| Out[&#x22;KubeMQ channel<br/>(out)&#x22;]

  class In,Out queue
  class Adapter,Producer external
  class Fn client
`"
/>

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

## Prerequisites [#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`:

<RunKubeMQ ports="[50000, 9090]" />

This guide targets &#x2A;*Java 17+**, &#x2A;*Spring Boot 3.2.0+**, and the **Spring Cloud 2023.0.0** release train.

## Add the Dependency [#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`.

<Tabs groupId="build-tool" items="['Gradle', 'Maven']">
  <Tab value="Gradle">
    ```kotlin title="build.gradle.kts"
    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")
    }
    ```
  </Tab>

  <Tab value="Maven">
    ```xml title="pom.xml"
    <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>
    ```
  </Tab>
</Tabs>

<Callout type="info">
  The KubeMQ multi-module build imports both `spring-boot-dependencies:3.2.0` and `spring-cloud-dependencies:2023.0.0`. Pair the Spring Cloud `2023.0.0` train with Spring Boot `3.2.x` — that combination is the one the binder is built and tested against.
</Callout>

## Functional Binding Configuration [#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`.

```yaml title="application.yml"
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 [#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:

```java title="KubeMQPattern.java"
public enum KubeMQPattern {
    EVENTS,
    EVENTS_STORE,
    QUEUES
}
```

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

```yaml title="application.yml"
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):

```yaml title="application.yml"
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 [#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.

<TypeTable
  type="{
  pattern: {
    description: &#x22;Messaging pattern for this binding: EVENTS, EVENTS_STORE, or QUEUES.&#x22;,
    type: &#x22;KubeMQPattern&#x22;,
    default: &#x22;EVENTS&#x22;,
  },
  eventsStoreType: {
    description: &#x22;Replay position for an EVENTS_STORE subscription (for example StartNewOnly, StartFromFirst, StartFromLast).&#x22;,
    type: &#x22;EventsStoreType&#x22;,
    default: &#x22;StartNewOnly&#x22;,
  },
  eventsStoreSequenceValue: {
    description: &#x22;Sequence number to start from when the events store type is sequence-based.&#x22;,
    type: &#x22;long&#x22;,
    default: &#x22;0&#x22;,
  },
  pollMaxMessages: {
    description: &#x22;Maximum number of messages to pull per poll for QUEUES bindings.&#x22;,
    type: &#x22;int&#x22;,
    default: &#x22;1&#x22;,
  },
  pollWaitTimeoutInSeconds: {
    description: &#x22;How long a QUEUES poll waits for messages before returning empty.&#x22;,
    type: &#x22;int&#x22;,
    default: &#x22;5&#x22;,
  },
  visibilitySeconds: {
    description: &#x22;Visibility window (seconds) during which a polled QUEUES message is hidden from other consumers while being processed.&#x22;,
    type: &#x22;int&#x22;,
    default: &#x22;30&#x22;,
  },
  autoAckMessages: {
    description: &#x22;When true, QUEUES messages are acknowledged automatically on receipt rather than after the function returns successfully.&#x22;,
    type: &#x22;boolean&#x22;,
    default: &#x22;false&#x22;,
  },
}"
/>

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

```yaml title="application.yml — Queues consumer tuning"
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 [#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.

| Adapter                                 | Pattern        | KubeMQ source                                                                                                                                                                      |
| --------------------------------------- | -------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |
| `KubeMQEventsMessageDrivenAdapter`      | `EVENTS`       | Subscribes via `PubSubClient.subscribeToEvents` — fire-and-forget pub/sub                                                                                                          |
| `KubeMQEventsStoreMessageDrivenAdapter` | `EVENTS_STORE` | Subscribes via `PubSubClient.subscribeToEventsStore` — persistent pub/sub with replay (`eventsStoreType` / `eventsStoreSequenceValue`)                                             |
| `KubeMQQueuesMessageDrivenAdapter`      | `QUEUES`       | Polls via `QueuesClient.receiveQueueMessages` — transactional receive; on a successful function return the message is acked, on failure it is nacked so another consumer can retry |

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 [#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 [#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:

```java title="KubeMQDestination.java"
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 [#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](#functional-binding-configuration).

<Steps>
  <Step>
    ### Define the Function [#define-the-function]

    The bean name `uppercase` matches the `uppercase-in-0` / `uppercase-out-0` binding names. The binder feeds each consumed message into this function and publishes whatever it returns.

    ```java title="ScsEventsFunctionConfig.java"
    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;
            };
        }
    }
    ```
  </Step>

  <Step>
    ### Send Into the Pipeline [#send-into-the-pipeline]

    To inject a message into the input binding programmatically, use Spring Cloud Stream's `StreamBridge`. Sending to `uppercase-in-0` publishes to the `spring-scs.events.in` channel, which the binder consumes and routes through the function.

    ```java title="ScsEventsRunner.java"
    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.");
        }
    }
    ```
  </Step>

  <Step>
    ### Run and Observe [#run-and-observe]

    Start the application (`web-application-type: none` makes it a console app). The runner publishes one message, the function upper-cases it, and the result is published to the output channel:

    ```text
    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.
    ```
  </Step>
</Steps>

### Queues and Events Store Variants [#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.

<Tabs groupId="scs-variant" items="['Queues', 'Events Store']">
  <Tab value="Queues">
    The `spring-cloud-stream-queues` example binds a `process` function over the `QUEUES` pattern. Queues are durable point-to-point channels, so the message survives until a consumer acks it.

    ```yaml title="application.yml"
    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
    ```

    ```java title="ScsQueuesFunctionConfig.java"
    @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;
        };
    }
    ```
  </Tab>

  <Tab value="Events Store">
    The `spring-cloud-stream-events-store` example binds a `transform` function over the `EVENTS_STORE` pattern. Events Store persists messages, so a consumer can replay history according to the binding's `eventsStoreType`.

    ```yaml title="application.yml"
    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
    ```

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

## When to Use the Binder [#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.

| Use the Spring Cloud Stream binder when…                                                         | Use `KubeMQTemplate` / annotations when…                                                     |
| ------------------------------------------------------------------------------------------------ | -------------------------------------------------------------------------------------------- |
| You are building functional, streaming pipelines from `Supplier` / `Function` / `Consumer` beans | You want explicit, imperative `send`/`listen` calls in your own services                     |
| You want destinations and binding wiring driven entirely by configuration                        | You want fine-grained control over each send and each subscription in code                   |
| You already use Spring Cloud Stream and want KubeMQ as the binder                                | You need patterns the binder does not cover, such as Commands and Queries (request/response) |
| You want to swap or compose binders (e.g. multi-binder topologies)                               | You are adding messaging to a conventional Spring Boot service                               |

<Callout type="info">
  The binder supports the **Events**, **Events Store**, and **Queues** patterns only. For synchronous request/response (Commands and Queries), use the [Commands & Queries](/integrations/spring-boot/how-to/commands-and-queries) API on `KubeMQTemplate` with the `@KubeMQCommandHandler` / `@KubeMQQueryHandler` annotations.
</Callout>

## Related [#related]

<Cards>
  <Card title="Getting Started" href="/integrations/spring-boot/tutorials/getting-started" description="Add the starter, configure a broker, and send and receive your first message." />

  <Card title="Events & Events Store" href="/integrations/spring-boot/how-to/events-and-events-store" description="Use the imperative template and listener API for fire-and-forget and persistent events." />

  <Card title="Queues" href="/integrations/spring-boot/how-to/queues" description="Durable point-to-point messaging with the template and @KubeMQQueueListener." />

  <Card title="Reference" href="/integrations/spring-boot/reference/api" description="The binder pattern/consumer properties, the KubeMQTemplate API, and listener annotations." />
</Cards>