StreamBridge's hashProducerProperties produces hash collisions across different binding names, causing Partition key cannot be null

[TmacZhou]

Issue: StreamBridge's hashProducerProperties produces hash collisions across different binding names, causing Partition key cannot be null

Metadata

• Component: spring-cloud-stream (core) — StreamBridge
• Version verified: 4.3.1 (also still present on main / 5.0.2)
• Binder: RabbitMQ (and any binder implementing ExtendedPropertiesBinder; see "Root cause")
• Type: Bug

Summary

StreamBridge.hashProducerProperties(...) computes a cache key for the shared FunctionInvocationWrapper used by StreamBridge.send(...). The hash does not reliably include the binding name, so two different producer bindings can produce the same hash and therefore share (reuse) the same cached FunctionInvocationWrapper. When one of those bindings is partitioned and the other is not, the non-partitioned binding can receive the partition handler/PartitionHandler of the partitioned one, and since its message has no partition key, it fails with:

java.lang.IllegalArgumentException: Partition key cannot be null

at PartitionHandler.extractKey(PartitionHandler.java:124).

Root cause

// StreamBridge.hashProducerProperties (identical in 4.3.1 and main/5.0.2)
private int hashProducerProperties(ProducerProperties producerProperties, String outputContentType) {
    int hash = outputContentType.hashCode()
            + Boolean.hashCode(producerProperties.isUseNativeEncoding())
            + Boolean.hashCode(producerProperties.isPartitioned())
            + producerProperties.getPartitionCount();

    if (producerProperties.getPartitionKeyExpression() != null && producerProperties.getBindingName() != null) {
        hash += producerProperties.getBindingName().hashCode();
    }

    return hash;
}

Two independent defects:

1. The binding name is only added when the binding is partitioned (guarded by getPartitionKeyExpression() != null). A non-partitioned binding never contributes its binding name to the hash. So a non-partitioned binding can collide with a partitioned binding that shares the same contentType, useNativeEncoding, isPartitioned-affecting fields, and partitionCount.

2. getBindingName() is actually null in the StreamBridge.send path even for partitioned bindings. populateBindingName(...) is only invoked inside BindingService.bindProducer(...) on the copy of the producer properties that is created when the binder is an ExtendedPropertiesBinder (e.g. RabbitMQ). StreamBridge.send calls BindingServiceProperties.getProducerProperties(bindingName) which returns the original ProducerProperties whose bindingName field is never populated. So the getBindingName() != null guard effectively prevents the binding name from ever being added in the send path.

Concrete collision (verified arithmetically)

With the same outputContentType, Boolean.hashCode(false) == 1237 and Boolean.hashCode(true) == 1231, so a partitioned binding and a non-partitioned binding collide whenever:

1231 + partitionCount_partitioned  ==  1237 + partitionCount_nonPartitioned

i.e. partitionCount_partitioned == partitionCount_nonPartitioned + 6. In our demo configuration, partitionedBinding-out-0 used partitionCount: 7 and nonPartitionedBinding-out-0 used the default partitionCount: 1 → 1231 + 7 == 1237 + 1 == 1238. Collision confirmed.

The shared cached FunctionInvocationWrapper then received the partition enhancer (via PartitionAwareFunctionWrapper.setEnhancer(...)), and the non-partitioned binding's message (which has no partition key) was processed by PartitionHandler, throwing Partition key cannot be null.

Workaround

Setting a partitionCount that cannot collide with any other binding's effective hash (here partition-count: 15 → 1237 + 15 = 1252, isolated from the partitioned binding's 1238) avoids the collision:

spring:
  cloud:
    stream:
      bindings:
        nonPartitionedBinding-out-0:
          producer:
            partition-count: 15

This is fragile: it depends on the hard-coded Boolean.hashCode values and requires every new StreamBridge.send binding to manually avoid collisions — easy to regress.

Suggestion

The cache key for StreamBridge should uniquely identify the binding. There are two complementary fixes:

1. Fix getBindingName() returning null in the send path. populateBindingName(...) is only invoked on the copy of ProducerProperties created inside BindingService.bindProducer(...) for ExtendedPropertiesBinder (e.g. RabbitMQ), while StreamBridge.send(...) reads the original ProducerProperties returned by BindingServiceProperties.getProducerProperties(...), whose bindingName is never populated. The framework should populate bindingName on the original object too (or ensure both paths share the same instance), so the existing getBindingName() != null guard can actually work.

2. Include the binding name unconditionally in the hash. Even with fix removed spring-xd-dirt dependencies #1, because the guard is getPartitionKeyExpression() != null && getBindingName() != null, a non-partitioned binding would still never contribute its binding name. The hash should be derived from the bindingName argument passed to StreamBridge.send("...", ...), which is always available, rather than relying on ProducerProperties#getBindingName(). For example:

private int hashProducerProperties(String bindingName, ProducerProperties producerProperties, String outputContentType) {
    int hash = outputContentType.hashCode()
            + bindingName.hashCode()
            + Boolean.hashCode(producerProperties.isUseNativeEncoding())
            + Boolean.hashCode(producerProperties.isPartitioned())
            + producerProperties.getPartitionCount();
    return hash;
}

With both fixes, each binding gets a distinct cache key — collisions (and the resulting Partition key cannot be null) are eliminated.

Reproduction

We can provide a minimal reproducer if needed. High-level steps:

1. Configure two outbound bindings through StreamBridge.send:
   • Binding A: partitioned (partitionKeyExpression set), partitionCount: 7.
   • Binding B: non-partitioned, default partitionCount: 1, same contentType.
2. Send a message to Binding A first (populates the cache), then send a message to Binding B.
3. Observe java.lang.IllegalArgumentException: Partition key cannot be null thrown from PartitionHandler.extractKey.

Environment

• Spring Cloud Stream 4.3.1 (Spring Cloud 2025.0.1), Spring Boot 3.5.x
• Also reproducible on main (5.0.2) — same code
• RabbitMQ binder, but the defect is reproducible with any binder that implements ExtendedPropertiesBinder