GH-3242: Key StreamBridge function cache by binding name

[kdelay]

Fixes #3242

What

StreamBridge.send(..) caches the FunctionInvocationWrapper under a hash of the producer properties. The binding name only entered that hash when partitionKeyExpression and ProducerProperties#getBindingName() were both non-null, and getBindingName() is never populated on the instance StreamBridge reads: BindingService#bindProducer calls populateBindingName(..) on the extended copy it builds for an ExtendedPropertiesBinder, while StreamBridge reads the original from BindingServiceProperties#getProducerProperties(bindingName).

So a partitioned binding can share its cached function with another binding. PartitionAwareFunctionWrapper sets the partition enhancer on that shared function and only clears it when the result carries no scst_partition header, so after a successful partitioned send the enhancer stays on the cached function. The next send on the colliding binding then runs through PartitionHandler without a partition key and fails with IllegalArgumentException: Partition key cannot be null.

The hash also summed its components, so Boolean.hashCode(true) + partitionCount and Boolean.hashCode(false) + partitionCount collide whenever the partitioned binding's count is 6 higher than the other one's.

Change

The cache is keyed by a record of outputContentType, useNativeEncoding, partitioned, partitionCount and bindingName instead of by a computed int, so ConcurrentHashMap distinguishes bindings by equality and no hash collision can make two of them share a function. The binding name comes from the send(..) argument, which is always available, and is only part of the key for partitioned bindings, so non-partitioned bindings keep sharing one cached function and the behaviour asserted by test_2783 is unchanged.

Verification

test_3242 in StreamBridgeTests reproduces the failure: a binding partitioned via partition-key-extractor-name (which leaves partitionKeyExpression null, so the old guard never added the binding name) with partition-count: 7, and a non-partitioned binding with partition-count: 1. On main the second send fails with Partition key cannot be null; with this change both sends succeed and only the partitioned message carries scst_partition.

partitionedBindingIsNotSharedWithHashCollidingBinding covers the collision case: two accepted configurations whose properties hash alike under Objects.hash(..) (both 682613285). A cache keyed by that hash hands the partition-aware function to the non-partitioned send, which then fails the same way; with the value key both sends succeed and the cache holds two entries.

• ./mvnw -f core/pom.xml clean install (the build CI runs): BUILD SUCCESS.
• StreamBridgeTests: 41 tests, 1 failure. The failure is test_3033, which fails the same way on main at a2bbc43, before this change (that module is currently commented out of the core reactor).
• Whole spring-cloud-stream-integration-tests module, this branch versus its base a2bbc43: 13 tests fail here and 14 on the base, and the set here is a subset of the base's. The one extra failure on the base (MultipleInputOutputFunctionTests.multiInputSingleOutputWithCustomContentType2) passes when run on its own, so it is flaky in the full run.

[HDPark95]

The binding-name source change fixes the reported 7/1 case, but the cache is still keyed by a single int, so the new code cannot provide the "must never share" guarantee in the comment.

I reproduced a remaining collision on this head with two accepted binding configurations:

partitioned dynamic binding: name="A>", partitionCount=120, partitionKeyExtractorName set
non-partitioned binding:      name="nonPartitioned-out-0", partitionCount=1

Objects.hash("application/json", false, true, 120, "A>") = 682613285
Objects.hash("application/json", false, false, 1, null)   = 682613285

The first send to A> succeeds and carries scst_partition. The following send to nonPartitioned-out-0 then reuses that cached wrapper and fails at PartitionHandler.extractKey with the same IllegalArgumentException: Partition key cannot be null reported in #3242. The focused StreamBridgeTests probe is 1 error on the submitted code.

Could the cache use a value key (for example, a record containing these five fields) instead of storing only the computed hash? ConcurrentHashMap would then use equality to distinguish colliding hashes. I verified that direction locally against the same probe: it becomes 1 test / 0 failures, and the existing test_2783 cache-sharing assertion remains 1 test / 0 failures.

[kdelay]

Confirmed on 25294be: Objects.hash("application/json", false, true, 120, "A>") and Objects.hash("application/json", false, false, 1, null) are both 682613285, and with that pair the send to nonPartitioned-out-0 fails at PartitionHandler.extractKey with Partition key cannot be null, exactly as reported in #3242.

Done in 76bf8f0. The cache now uses a value key: streamBridgeFunctionCache is keyed by a record of outputContentType, useNativeEncoding, partitioned, partitionCount and bindingName, so ConcurrentHashMap tells the two bindings apart by equality and a hash collision can no longer make them share a function. The hashProducerProperties helper is gone. Your A>/120 versus nonPartitioned-out-0/1 pair is now in the suite as partitionedBindingIsNotSharedWithHashCollidingBinding; besides the two sends it asserts the cache holds two entries.

I also rebased onto main (a2bbc43) since the branch had fallen behind.

Verified on 76bf8f0:

• ./mvnw -f core/pom.xml clean install: BUILD SUCCESS.
• StreamBridgeTests: 41 tests, 1 failure. The failure is test_3033, which fails the same way on main at a2bbc43, before this change.
• test_2783 (cache sharing) and test_3105 pass unchanged.