Make BindingServiceProperties.bindings thread-safe and case-insensitive

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<String, BindingProperties> 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<String, BindingProperties> bindings = new ConcurrentSkipListMap<>(String.CASE_INSENSITIVE_ORDER);
 
 	/**
 	 * Additional per-binder properties (see {@link BinderProperties}) if more than one

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<String> 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<Future<?>> futures = new ArrayList<>();
+				for (int thread = 0; thread < 16; thread++) {
+					List<String> 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();
+		}
+	}
+
 }