SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-2299 Add test to ensure scst_partition header is always set in concurrent scenarios

Browse Source
This commit is contained in:
Sébastien NUSSBAUMER
2022-03-14 09:47:34 +01:00
committed by Oleg Zhurakousky
parent 8a8a440c73
commit 9f28bcde3d
1 changed files with 44 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/function/StreamBridgeTests.java
View File

@@ -18,6 +18,7 @@ package org.springframework.cloud.stream.function;
import java.lang.reflect.Field;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
@@ -26,6 +27,8 @@ import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -107,6 +110,47 @@ public class StreamBridgeTests {
}
}
/*
* This test verifies that when a partition key expression is set, then scst_partition is always set, even in
* concurrent scenarios.
* See https://github.com/spring-cloud/spring-cloud-stream/issues/2299 for more details
*/
@Test
void test_2299_scstPartitionAlwaysSetEvenInConcurrentScenarios() throws Exception {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(EmptyConfiguration.class)).web(
WebApplicationType.NONE).run("--spring.cloud.stream.source=outputA",
"--spring.cloud.stream.bindings.outputA-out-0.producer.partition-count=3",
"--spring.cloud.stream.bindings.outputA-out-0.producer.partition-key-expression=headers['partitionKey']",
"--spring.jmx.enabled=false")) {
StreamBridge streamBridge = context.getBean(StreamBridge.class);
int threadCount = 10;
Set<Thread> threads = IntStream.range(0, threadCount)
.mapToObj(i -> (Runnable) () -> IntStream.range(0, 100).forEach(j -> {
String value = "M-" + i + "-" + j;
streamBridge.send("outputA-out-0",
MessageBuilder.withPayload(value).setHeader("partitionKey", value).build());
})).map(Thread::new).collect(Collectors.toSet());
threads.forEach(Thread::start);
for (Thread thread : threads) {
thread.join();
}
int messagesWithoutScstPartition = 0;
OutputDestination output = context.getBean(OutputDestination.class);
Message<byte[]> message = output.receive(1000, "outputA-out-0");
while (message != null) {
if (!message.getHeaders().containsKey("scst_partition")) {
messagesWithoutScstPartition++;
}
message = output.receive(1000, "outputA-out-0");
}
assertThat(messagesWithoutScstPartition).isEqualTo(0);
}
}
@Test
void testWithOutputContentTypeWildCardBindings() throws Exception {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(TestChannelBinderConfiguration