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(); + } + } + }