Skip to content
JavaAgentic

Type at least two characters. Try “RAG”, “pgvector” or “tool calling”.

Spring Cloud Stream & Message-Driven Services

One programming model over Kafka and RabbitMQ: functional bindings, destination configuration, per-binder tuning, dead-letter handling and the in-memory test binder.

Intermediate6 min readUpdated
On this page

Spring Cloud Stream puts one programming model over several brokers. You write a Function or a Consumer; the framework handles connection, serialisation, consumer groups, retries and dead-letter routing.

Key Takeaways

  • The unit of work is a java.util.function bean — no broker API in your code.
  • Binding names derive from bean name plus suffix, which is the usual reason nothing binds.
  • The binder supplies broker specifics; swapping it changes the dependency, not the code.
  • Always enable the DLQ and bound the retries.
  • The test binder runs the whole pipeline in memory with no broker.

The functional model

OrderFunctions.java
@Configuration
public class OrderFunctions {
 
    /** Consumer: one input, no output. Binds as processOrder-in-0. */
    @Bean
    public Consumer<Message<OrderPlaced>> processOrder(FulfilmentService fulfilment) {
        return message -> {
            OrderPlaced event = message.getPayload();
            String eventId = message.getHeaders().get("eventId", String.class);
            // At-least-once delivery: deduplicate before doing anything.
            if (!processed.record(eventId)) return;
            fulfilment.begin(event);
        };
    }
 
    /** Function: input and output. Binds as enrichOrder-in-0 / enrichOrder-out-0. */
    @Bean
    public Function<OrderPlaced, EnrichedOrder> enrichOrder(CustomerClient customers) {
        return order -> order.enrichedWith(customers.fetch(order.customerId()));
    }
 
    /** Supplier: polled on a schedule, or triggered by StreamBridge. */
    @Bean
    public Supplier<HealthPing> heartbeat() {
        return () -> new HealthPing(Instant.now());
    }
 
    /** Splitting one input across several outputs. */
    @Bean
    public Function<OrderPlaced, Tuple2<Flux<HighValueOrder>, Flux<StandardOrder>>> routeByValue() {
        return flux -> Tuples.of(
                Flux.from(flux).filter(o -> o.totalMinorUnits() > 100_000).map(HighValueOrder::from),
                Flux.from(flux).filter(o -> o.totalMinorUnits() <= 100_000).map(StandardOrder::from));
    }
}

There is no @KafkaListener, no @RabbitListener, no template. The function is ordinary Java that can be unit-tested by calling it directly.

Binding configuration

application.yml
spring:
  cloud:
    function:
      # Explicit. Relying on bean-name inference is where the silent
      # "nothing is bound and nothing complains" failure comes from.
      definition: 'processOrder;enrichOrder'
    stream:
      bindings:
        processOrder-in-0:
          destination: orders
          group: fulfilment              # the consumer group — required for scaling
          consumer:
            max-attempts: 3
            back-off-initial-interval: 1000
            back-off-multiplier: 2.0
            concurrency: 3
        enrichOrder-in-0:
          destination: orders
          group: enrichment
        enrichOrder-out-0:
          destination: orders-enriched
          producer:
            partition-key-expression: payload.customerId
            partition-count: 6
 
      kafka:
        binder:
          brokers: 'kafka-1:9092,kafka-2:9092'
          required-acks: all
          configuration:
            enable.idempotence: true
            compression.type: zstd
        bindings:
          processOrder-in-0:
            consumer:
              enable-dlq: true
              dlq-name: orders.dlq
              dlq-partitions: 1
              auto-commit-on-error: false
              start-offset: earliest

The group property is what turns a broadcast into a competing-consumer queue. Without it, every instance of your service receives every message — which is occasionally what you want and usually a bug that only shows up once you scale past one replica.

concurrency sitting next to it deserves a second look. On Kafka it creates that many consumer threads inside the instance, each claiming whole partitions, so concurrency above the partition count buys nothing at all — three instances at concurrency 3 need at least nine partitions to use what you asked for. It also interacts with ordering: records stay ordered within a partition, so raising concurrency never reorders a single key, but two keys that previously shared a thread now advance independently. On RabbitMQ the same setting creates concurrent consumers on one queue, where there is no partition preserving order in the first place.

Serialisation is worth pinning explicitly too. The default content type is application/json, and the framework converts the payload to your function's parameter type using whichever message converter matches. That is pleasant until a producer changes shape, at which point the failure surfaces as a conversion error deep inside the binder rather than as a contract violation. Setting content-type per binding — and using a schema registry with Avro or Protobuf where the contract genuinely matters — turns a runtime surprise into something caught before deployment.

Where the abstraction leaks

The programming model is portable. The delivery semantics underneath are not, and code written against one broker's guarantees will not behave identically on the other.

Be clear-eyed about what portability buys you. It removes boilerplate and lets one team's idioms apply across services using different brokers. It does not make the broker choice reversible: a consumer that relies on replaying from an offset has no equivalent on RabbitMQ, and one that relies on routing keys has no equivalent on Kafka.

Treat the binder as a way to write less code, not as an escape from understanding the broker.

Producing outside a function

StreamBridgeUsage.java
@Service
public class OrderService {
 
    private final StreamBridge bridge;
 
    @Transactional
    public Order place(PlaceOrderCommand command) {
        Order order = repository.save(command.toOrder());
 
        // For an event triggered by a request rather than by another message.
        // Note: this is NOT transactional with the database write — use an
        // outbox if the event must not be lost.
        bridge.send("orderPlaced-out-0", MessageBuilder
                .withPayload(new OrderPlaced(order.id(), order.customerId()))
                .setHeader("eventId", UUID.randomUUID().toString())
                .setHeader(KafkaHeaders.KEY, order.id())
                .build());
 
        return order;
    }
}

That comment is the important part. StreamBridge.send is not enrolled in the database transaction, so a rollback after a successful send leaves an event describing something that never happened. Where that matters, write to an outbox table inside the transaction and let a relay publish.

Error handling

ErrorHandling.java
@Configuration
public class StreamErrorConfig {
 
    /** Per-binding error channel: <destination>.<group>.errors */
    @ServiceActivator(inputChannel = "orders.fulfilment.errors")
    public void handleOrderError(ErrorMessage message) {
        var original = (Message<?>) message.getOriginalMessage();
        log.error("failed to process {}", original != null ? original.getPayload() : "unknown",
                  message.getPayload());
        alerting.raise("stream-error", "fulfilment");
    }
 
    /** Global fallback for bindings without a specific handler. */
    @ServiceActivator(inputChannel = "errorChannel")
    public void handleGlobalError(ErrorMessage message) {
        log.error("unhandled stream error", message.getPayload());
    }
}

The retry configured under consumer.max-attempts is in-process and blocking, so a long backoff holds the consumer thread. Once retries are exhausted, the binder's DLQ takes the message — provided you enabled it. Without enable-dlq, an unrecoverable message is retried forever and blocks everything behind it on that partition.

Testing without a broker

OrderFunctionsTest.java
@SpringBootTest
@Import(TestChannelBinderConfiguration.class)
class OrderFunctionsTest {
 
    @Autowired InputDestination input;
    @Autowired OutputDestination output;
 
    @Test
    void enrichesAnOrderWithCustomerData() {
        input.send(MessageBuilder.withPayload(new OrderPlaced("ORD-1", "cus_1")).build(),
                   "orders");
 
        Message<byte[]> result = output.receive(2000, "orders-enriched");
 
        assertThat(result).isNotNull();
        assertThat(new String(result.getPayload())).contains("PREMIUM");
    }
}

The test binder replaces Kafka or RabbitMQ with in-memory channels, so the whole pipeline — binding, serialisation, the function itself — runs in milliseconds with no Docker. Keep a smaller set of Testcontainers tests for the binder-specific behaviour the in-memory version cannot reproduce, such as DLQ routing and rebalancing.

What to take away

Write plain functions and let bindings connect them. Set spring.cloud.function.definition explicitly and always give consumers a group. Enable the DLQ with bounded retries. Use the test binder for fast feedback and real containers for the parts that depend on broker semantics — and remember those semantics are not portable even though your code is.

Frequently Asked Questions

Is the broker really swappable?
The programming model is. The semantics are not — Kafka gives you replay and partition ordering, RabbitMQ gives you routing and per-message acknowledgement, and code written against one broker's guarantees will not behave the same on the other. Treat the abstraction as removing boilerplate, not as making the choice reversible.
Why is my function not bound to a destination?
Spring Cloud Stream derives binding names from the bean name plus a suffix, so a bean named processOrder binds as processOrder-in-0 and processOrder-out-0. If your configuration uses a different name, nothing binds and no error is raised. Set spring.cloud.function.definition explicitly.
How do I handle a poison message?
Enable the binder DLQ — enableDlq for Kafka, autoBindDlq for RabbitMQ — and set a bounded retry count. Without a DLQ, a message that always fails is retried forever and blocks its partition or queue, turning one bad record into an outage.

Related tutorials