GH-3242: Key StreamBridge function cache by binding name - #3244
Open
kdelay wants to merge 1 commit into
Open
Conversation
StreamBridge caches the FunctionInvocationWrapper used by send(..) under a hash of the producer properties. The binding name was only part of that hash when partitionKeyExpression and ProducerProperties#getBindingName() were both set, and getBindingName() is never populated on the instance StreamBridge reads: BindingService#bindProducer populates it on the extended copy it creates for an ExtendedPropertiesBinder, not on the original returned by BindingServiceProperties#getProducerProperties. A partitioned binding therefore shares its cached function with any other binding whose properties hash to the same value, and since the partition enhancer is left on the cached function after a send that produced a partition header, the next send on the colliding binding fails with IllegalArgumentException: Partition key cannot be null. Take the binding name from the send(..) argument, which is always available, and include it for partitioned bindings. Non-partitioned bindings keep sharing a cached function as before. Use Objects.hash so that the components are mixed rather than summed, which is what let a partitioned binding with partitionCount N collide with a non-partitioned one with partitionCount N-6. Signed-off-by: kdelay <kdelay20@gmail.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #3242
What
StreamBridge.send(..)caches theFunctionInvocationWrapperunder a hash of the producer properties. The binding name only entered that hash whenpartitionKeyExpressionandProducerProperties#getBindingName()were both non-null, andgetBindingName()is never populated on the instanceStreamBridgereads:BindingService#bindProducercallspopulateBindingName(..)on the extended copy it builds for anExtendedPropertiesBinder, whileStreamBridgereads the original fromBindingServiceProperties#getProducerProperties(bindingName).So a partitioned binding can share its cached function with another binding.
PartitionAwareFunctionWrappersets the partition enhancer on that shared function and only clears it when the result carries noscst_partitionheader, so after a successful partitioned send the enhancer stays on the cached function. The next send on the colliding binding then runs throughPartitionHandlerwithout a partition key and fails withIllegalArgumentException: Partition key cannot be null.The hash also summed its components, so
Boolean.hashCode(true) + partitionCountandBoolean.hashCode(false) + partitionCountcollide whenever the partitioned binding's count is 6 higher than the other one's.Change
hashProducerPropertiesnow takes the binding name from thesend(..)argument, which is always available, and includes it for partitioned bindings. Non-partitioned bindings keep sharing one cached function, so the behaviour asserted bytest_2783is unchanged.Objects.hashreplaces the summation so the components are mixed.Verification
test_3242inStreamBridgeTestsreproduces the failure: a binding partitioned viapartition-key-extractor-name(which leavespartitionKeyExpressionnull, so the old guard never added the binding name) withpartition-count: 7, and a non-partitioned binding withpartition-count: 1. Onmainthe second send fails withPartition key cannot be null; with this change both sends succeed and only the partitioned message carriesscst_partition../mvnw -f core/pom.xml clean install(the build CI runs): BUILD SUCCESS.StreamBridgeTests: 39/40 pass.test_3033fails onmainas well, before this change, and is unrelated (that module is currently commented out of thecorereactor).spring-cloud-stream-integration-testsmodule: the set of failing tests is identical before and after this change.