From 5d66d26a622ff41019a371276ab62dd8b18e3927 Mon Sep 17 00:00:00 2001 From: Akash Bhogar <55949790+AkashB23@users.noreply.github.com> Date: Mon, 2 Nov 2026 15:00:16 +0530 Subject: [PATCH] Make BindingServiceProperties.bindings thread-safe and case-insensitive GH-3158 changed the bindings map back to a TreeMap with a case-insensitive comparator to restore case-insensitive binding names. That also dropped the thread safety added by gh-3011 and kept by gh-3076. The map is mutated on read: getBindingDestination(), getBindingProperties(), getProducerProperties() and getConsumerProperties() add a missing binding on first use (bindIfNecessary -> bindToDefault -> bindings.put). StreamBridge, BindingService and the binders call these from many threads without any lock. Concurrent first use of binding names corrupts the red-black tree: entries are lost, size() no longer matches the iteration, a NullPointerException is thrown from TreeMap.rotateRight (the failure reported in gh-3011) or a thread spins forever in TreeMap.successor while holding whatever monitor its caller holds, which stalls every thread that needs the bindings. Use a ConcurrentSkipListMap with String.CASE_INSENSITIVE_ORDER. It keeps what GH-3158 needs (case-insensitive lookup, original key casing, sorted iteration, same as the TreeMap) and is safe for concurrent use. A ConcurrentHashMap cannot be used because it is case-sensitive. The ConcurrentSkipListMap of gh-3011 was replaced by gh-3076 because of slower start-up. That issue's thread does not show the map as the cause: the reporter saw no improvement with the ConcurrentHashMap build and traced the slowdown to a Spring Boot change in how child contexts are created, plus binding properties placed under spring.cloud.config instead of spring.cloud.stream, which made every first getBinder() call take the bindToDefault() path. The new test, in BindingServiceTests, fails on the TreeMap (entries are lost or the map throws) and passes with this change. Fixes gh-3321 See gh-3011, gh-3076, gh-3158 Signed-off-by: Akash Bhogar <55949790+AkashB23@users.noreply.github.com> --- .../config/BindingServiceProperties.java | 6 ++- .../stream/binding/BindingServiceTests.java | 49 +++++++++++++++++++ 2 files changed, 53 insertions(+), 2 deletions(-) diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java index e4bda976c5..f3a161c44b 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/config/BindingServiceProperties.java @@ -20,7 +20,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; -import java.util.TreeMap; +import java.util.concurrent.ConcurrentSkipListMap; import com.fasterxml.jackson.annotation.JsonInclude; import com.fasterxml.jackson.annotation.JsonInclude.Include; @@ -106,7 +106,9 @@ public class BindingServiceProperties * For example; This sets the content-type for the 'input' binding of a Sink * application: 'spring.cloud.stream.bindings.input.contentType=text/plain' */ - private Map bindings = new TreeMap<>(String.CASE_INSENSITIVE_ORDER); + // Case-insensitive (GH-3158) and thread-safe: getBindingProperties() and friends add + // missing bindings on first use and are called concurrently (e.g., by StreamBridge). + private Map bindings = new ConcurrentSkipListMap<>(String.CASE_INSENSITIVE_ORDER); /** * Additional per-binder properties (see {@link BinderProperties}) if more than one diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java index 0476d6d710..c8748bff7d 100644 --- a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java +++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/BindingServiceTests.java @@ -16,11 +16,21 @@ package org.springframework.cloud.stream.binding; +import java.util.ArrayList; +import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; +import java.util.Random; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; import tools.jackson.databind.ObjectMapper; import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.Timeout; import org.mockito.InOrder; import org.mockito.Mockito; @@ -30,6 +40,7 @@ import org.springframework.cloud.stream.binder.ConsumerProperties; import org.springframework.cloud.stream.config.BindingProperties; import org.springframework.cloud.stream.config.BindingServiceProperties; +import org.springframework.context.support.GenericApplicationContext; import org.springframework.integration.channel.DirectChannel; import org.springframework.messaging.MessageChannel; @@ -79,4 +90,42 @@ void unbindsConsumerBindingsIfLaterConsumerBindingFails() { assertThat(service.getConsumerBindings(inputChannelName)).isEmpty(); } + @Test + @Timeout(120) + void concurrentFirstUseOfBindingsKeepsTheMapConsistent() throws Exception { + List names = new ArrayList<>(); + for (int i = 0; i < 1000; i++) { + names.add("binding-" + i); + } + ExecutorService executor = Executors.newFixedThreadPool(16); + try { + for (int round = 0; round < 5; round++) { + GenericApplicationContext context = new GenericApplicationContext(); + context.refresh(); + BindingServiceProperties properties = new BindingServiceProperties(); + properties.setApplicationContext(context); + + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(); + for (int thread = 0; thread < 16; thread++) { + List shuffled = new ArrayList<>(names); + Collections.shuffle(shuffled, new Random(thread)); + futures.add(executor.submit(() -> { + start.await(); + shuffled.forEach(properties::getBindingProperties); + return null; + })); + } + start.countDown(); + for (Future future : futures) { + future.get(30, TimeUnit.SECONDS); + } + assertThat(properties.getBindings().keySet()).containsExactlyInAnyOrderElementsOf(names); + } + } + finally { + executor.shutdownNow(); + } + } + }