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

GH-2923: StreamBridge and Kafka send with virtual threads

Browse Source 
* Sending Kafka events via StreamBridge#send using virtual threads causes pinning. 
     Use a ReentrantLock to avoid pinning.
This commit is contained in:
Michał Rowicki
2024-03-19 21:40:10 +01:00
committed by GitHub
parent e80f53fbe8
commit f836e1cf54
1 changed files with 53 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files

96
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/function/StreamBridge.java
View File

@@ -24,6 +24,7 @@ import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import io.micrometer.context.ContextExecutorService;
@@ -83,6 +84,7 @@ import org.springframework.util.StringUtils;
* @author Oleg Zhurakousky
* @author Soby Chacko
* @author Byungjun You
* @author Michał Rowicki
* @since 3.0.3
*
*/
@@ -118,6 +120,8 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
private static final boolean isContextPropagationPresent = ClassUtils.isPresent(
"io.micrometer.context.ContextSnapshotFactory", StreamBridge.class.getClassLoader());
private static final ReentrantLock lock = new ReentrantLock();
/**
*
* @param functionCatalog instance of {@link FunctionCatalog}
@@ -249,56 +253,62 @@ public final class StreamBridge implements StreamOperations, SmartInitializingSi
}
@SuppressWarnings({ "unchecked"})
synchronized MessageChannel resolveDestination(String destinationName, ProducerProperties producerProperties, String binderName) {
MessageChannel messageChannel = null;
if (StringUtils.hasText(binderName)) {
messageChannel = this.channelCache.get(binderName + ":" + destinationName);
}
else {
messageChannel = this.channelCache.get(destinationName);
}
if (messageChannel == null) {
if (this.applicationContext.containsBean(destinationName)) {
messageChannel = this.applicationContext.getBean(destinationName, MessageChannel.class);
String[] consumerBindingNames = this.bindingService.getConsumerBindingNames();
if (messageChannel instanceof AbstractMessageChannel) {
addPartitioningInterceptorIfNeedBe(producerProperties, destinationName, (AbstractMessageChannel) messageChannel);
}
if (ObjectUtils.containsElement(consumerBindingNames, destinationName)) { //GH-2563
logger.warn("You seem to be sending data to the input binding. It is not "
+ "recommended, since you are bypassing the binder and this the messaging system exposed by the binder.");
}
MessageChannel resolveDestination(String destinationName, ProducerProperties producerProperties, String binderName) {
lock.lock();
try {
MessageChannel messageChannel = null;
if (StringUtils.hasText(binderName)) {
messageChannel = this.channelCache.get(binderName + ":" + destinationName);
}
else {
messageChannel = this.isAsync() ? new ExecutorChannel(this.executorService) : new DirectWithAttributesChannel();
((AbstractSubscribableChannel) messageChannel).setApplicationContext(applicationContext);
((AbstractSubscribableChannel) messageChannel).setComponentName(destinationName);
if (this.destinationBindingCallback != null) {
Object extendedProducerProperties = this.bindingService
.getExtendedProducerProperties(messageChannel, destinationName);
this.destinationBindingCallback.configure(destinationName, messageChannel,
producerProperties, extendedProducerProperties);
}
Binder binder = null;
if (StringUtils.hasText(binderName)) {
BinderFactory binderFactory = this.applicationContext.getBean(BinderFactory.class);
binder = binderFactory.getBinder(binderName, messageChannel.getClass());
}
addPartitioningInterceptorIfNeedBe(producerProperties, destinationName, (AbstractMessageChannel) messageChannel);
addGlobalChannelInterceptorProcessor((AbstractMessageChannel) messageChannel, destinationName);
this.bindingService.bindProducer(messageChannel, destinationName, true, binder);
if (StringUtils.hasText(binderName)) {
this.channelCache.put(binderName + ":" + destinationName, messageChannel);
messageChannel = this.channelCache.get(destinationName);
}
if (messageChannel == null) {
if (this.applicationContext.containsBean(destinationName)) {
messageChannel = this.applicationContext.getBean(destinationName, MessageChannel.class);
String[] consumerBindingNames = this.bindingService.getConsumerBindingNames();
if (messageChannel instanceof AbstractMessageChannel) {
addPartitioningInterceptorIfNeedBe(producerProperties, destinationName, (AbstractMessageChannel) messageChannel);
}
if (ObjectUtils.containsElement(consumerBindingNames, destinationName)) { //GH-2563
logger.warn("You seem to be sending data to the input binding. It is not "
+ "recommended, since you are bypassing the binder and this the messaging system exposed by the binder.");
}
}
else {
this.channelCache.put(destinationName, messageChannel);
messageChannel = this.isAsync() ? new ExecutorChannel(this.executorService) : new DirectWithAttributesChannel();
((AbstractSubscribableChannel) messageChannel).setApplicationContext(applicationContext);
((AbstractSubscribableChannel) messageChannel).setComponentName(destinationName);
if (this.destinationBindingCallback != null) {
Object extendedProducerProperties = this.bindingService
.getExtendedProducerProperties(messageChannel, destinationName);
this.destinationBindingCallback.configure(destinationName, messageChannel,
producerProperties, extendedProducerProperties);
}
Binder binder = null;
if (StringUtils.hasText(binderName)) {
BinderFactory binderFactory = this.applicationContext.getBean(BinderFactory.class);
binder = binderFactory.getBinder(binderName, messageChannel.getClass());
}
addPartitioningInterceptorIfNeedBe(producerProperties, destinationName, (AbstractMessageChannel) messageChannel);
addGlobalChannelInterceptorProcessor((AbstractMessageChannel) messageChannel, destinationName);
this.bindingService.bindProducer(messageChannel, destinationName, true, binder);
if (StringUtils.hasText(binderName)) {
this.channelCache.put(binderName + ":" + destinationName, messageChannel);
}
else {
this.channelCache.put(destinationName, messageChannel);
}
}
}
}
return messageChannel;
return messageChannel;
}
finally {
lock.unlock();
}
}
private void addPartitioningInterceptorIfNeedBe(ProducerProperties producerProperties, String destinationName, AbstractMessageChannel messageChannel) {