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

GH-2512, 2507 Establish binder identity & change error channel binding scheme

Browse Source 
This commit establishes the concept of binder identity (binder instance identity) to be used in cases where unique-to-the-binder-instance naming is required.
For example such naming is required to fix GH-2507 where error channel names could colide if two binders use the same destination name.

Resolves #2512
Resolves #2507
This commit is contained in:
Oleg Zhurakousky
2022-09-27 16:02:24 +02:00
parent 65c6274a11
commit 5035a40990
8 changed files with 97 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -374,6 +374,11 @@ public class KafkaMessageChannelBinder extends
return this.transactionManager == null ? null : this.transactionManager.getProducerFactory();
}
@Override
public String getBinderIdentity() {
return "kafka-" + super.getBinderIdentity();
}
@Override
protected MessageHandler createProducerMessageHandler(
final ProducerDestination destination,

10
binders/kafka-binder/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -1039,6 +1039,7 @@ public class KafkaBinderTests extends
consumerProperties.setMultiplex(true);
consumerProperties.getExtension().setDlqPartitions(dlqPartitions);
consumerProperties.setConcurrency(2);
consumerProperties.populateBindingName("foobar");
DirectChannel moduleInputChannel = createBindableChannel("input",
createConsumerBindingProperties(consumerProperties));
@@ -1091,8 +1092,9 @@ public class KafkaBinderTests extends
ApplicationContext context = TestUtils.getPropertyValue(binder.getBinder(),
"applicationContext", ApplicationContext.class);
SubscribableChannel boundErrorChannel = context
.getBean(consumerDest + ".testGroup.errors-0", SubscribableChannel.class);
SubscribableChannel boundErrorChannel = context.getBean(binder.getBinder().getBinderIdentity() + ".foobar.errors-0", SubscribableChannel.class);
// SubscribableChannel boundErrorChannel = context
// .getBean(consumerDest + ".errors-0", SubscribableChannel.class);
SubscribableChannel globalErrorChannel = context.getBean("errorChannel",
SubscribableChannel.class);
final AtomicReference<Message<?>> boundErrorChannelMessage = new AtomicReference<>();
@@ -2450,12 +2452,12 @@ public class KafkaBinderTests extends
new KafkaProducerProperties());
producerProps.setHeaderMode(HeaderMode.none);
producerProps.setErrorChannelEnabled(true);
producerProps.populateBindingName("foobar");
Binding<MessageChannel> producerBinding = binder.bindProducer("ec.0",
moduleOutputChannel, producerProps);
final Message<?> message = MessageBuilder.withPayload("bad")
.setHeader(MessageHeaders.CONTENT_TYPE, "application/json").build();
SubscribableChannel ec = binder.getApplicationContext().getBean("ec.0.errors",
SubscribableChannel.class);
SubscribableChannel ec = binder.getApplicationContext().getBean(binder.getBinder().getBinderIdentity() + ".foobar.errors", SubscribableChannel.class);
final AtomicReference<Message<?>> errorMessage = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(2);
ec.subscribe(message1 -> {

15
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/main/java/org/springframework/cloud/stream/binder/rabbit/RabbitMessageChannelBinder.java
View File

@@ -306,6 +306,11 @@ public class RabbitMessageChannelBinder extends
return this.extendedBindingProperties.getExtendedPropertiesEntryClass();
}
@Override
public String getBinderIdentity() {
return "rabbit-" + super.getBinderIdentity();
}
@Override
protected MessageHandler createProducerMessageHandler(
final ProducerDestination producerDestination,
@@ -952,11 +957,11 @@ public class RabbitMessageChannelBinder extends
};
}
@Override
protected String errorsBaseName(ConsumerDestination destination, String group,
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
return destination.getName() + ".errors";
}
// @Override
// protected String errorsBaseName(ConsumerDestination destination, String group,
// ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties) {
// return destination.getName() + ".errors";
// }
private String deadLetterExchangeName(RabbitCommonProperties properties) {
if (properties.getDeadLetterExchange() == null) {

10
binders/rabbit-binder/spring-cloud-stream-binder-rabbit/src/test/java/org/springframework/cloud/stream/binder/rabbit/RabbitBinderTests.java
View File

@@ -260,11 +260,14 @@ public class RabbitBinderTests extends
new BindingProperties());
ExtendedProducerProperties<RabbitProducerProperties> producerProps = createProducerProperties(testInfo);
producerProps.setErrorChannelEnabled(true);
producerProps.populateBindingName("output");
Binding<MessageChannel> producerBinding = binder.bindProducer("ec.0",
moduleOutputChannel, producerProps);
final Message<?> message = MessageBuilder.withPayload("bad".getBytes())
.setHeader(MessageHeaders.CONTENT_TYPE, "foo/bar").build();
SubscribableChannel ec = binder.getApplicationContext().getBean("ec.0.errors",
String s = testBinder.getBinder().getBinderIdentity() + ".output.errors";
//return this.getBinderIdentity() + "-" + this.hashCode() + "." + bindingName + ".errors";
SubscribableChannel ec = binder.getApplicationContext().getBean(s,
SubscribableChannel.class);
final AtomicReference<Message<?>> errorMessage = new AtomicReference<>();
final CountDownLatch latch = new CountDownLatch(2);
@@ -1299,6 +1302,7 @@ public class RabbitBinderTests extends
properties.setMaxAttempts(withRetry ? 2 : 1);
properties.setPartitioned(true);
properties.setInstanceIndex(0);
properties.populateBindingName("blahblah");
DirectChannel input0 = createBindableChannel("input",
createConsumerBindingProperties(properties));
input0.setBeanName("test.input0DLQ");
@@ -1355,11 +1359,9 @@ public class RabbitBinderTests extends
}
});
ApplicationContext context = TestUtils.getPropertyValue(binder.getBinder(),
"applicationContext", ApplicationContext.class);
SubscribableChannel boundErrorChannel = context.getBean(
"bindertest.partPubDLQ.0.dlqPartGrp-0.errors", SubscribableChannel.class);
SubscribableChannel boundErrorChannel = context.getBean(testBinder.getBinder().getBinderIdentity() + ".blahblah.errors", SubscribableChannel.class);
SubscribableChannel globalErrorChannel = context.getBean("errorChannel",
SubscribableChannel.class);
final AtomicReference<Message<?>> boundErrorChannelMessage = new AtomicReference<>();

16
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractBinder.java
View File

@@ -26,6 +26,7 @@ import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
import org.springframework.cloud.stream.annotation.StreamRetryTemplate;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.support.AbstractApplicationContext;
@@ -166,6 +167,21 @@ public abstract class AbstractBinder<T, C extends ConsumerProperties, P extends
+ (StringUtils.hasText(group) ? group : "default");
}
/**
* Attempts to get {@link BindingServiceProperties} from application context.
*
* @return instance of {@link BindingServiceProperties} or null.
*/
protected BindingServiceProperties getBindingServiceProperties() {
try {
return getApplicationContext().getBean(BindingServiceProperties.class);
}
catch (Exception e) {
// ignore
return null;
}
}
/**
* Create and configure a default retry template unless one has already been provided
* via @Bean by an application.

66
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/AbstractMessageChannelBinder.java
View File

@@ -251,8 +251,18 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
try {
producerDestination = this.provisioningProvider
.provisionProducerDestination(destination, producerProperties);
SubscribableChannel errorChannel = producerProperties.isErrorChannelEnabled()
? registerErrorInfrastructure(producerDestination) : null;
BindingProperties bp = null;
BindingServiceProperties bsp = this.getBindingServiceProperties();
if (bsp != null) {
String bindingName = StringUtils.hasText(producerProperties.getBindingName()) ? producerProperties.getBindingName() : destination;
bp = bsp.getBindingProperties(bindingName);
}
SubscribableChannel errorChannel = (bp != null && StringUtils.hasText(bp.getErrorHandlerDefinition())) || producerProperties.isErrorChannelEnabled()
? registerErrorInfrastructure(producerDestination, producerProperties.getBindingName()) : null;
producerMessageHandler = createProducerMessageHandler(producerDestination,
producerProperties, outputChannel, errorChannel);
customizeProducerMessageHandler(producerMessageHandler, producerDestination.getName());
@@ -314,7 +324,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
@Override
public void afterUnbind() {
try {
destroyErrorInfrastructure(producerDestination);
destroyErrorInfrastructure(producerDestination, producerProperties.getBindingName());
final ReactiveStreamsConsumer rsc = reactiveStreamsConsumerRef.get();
if (rsc != null && rsc.isRunning()) {
rsc.destroy();
@@ -636,9 +646,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
* @return the channel.
*/
private SubscribableChannel registerErrorInfrastructure(
ProducerDestination destination) {
ProducerDestination destination, String bindingName) {
String errorChannelName = errorsBaseName(destination);
String errorChannelName = errorsBaseName(destination, bindingName);
SubscribableChannel errorChannel = new PublishSubscribeChannel();
if (getApplicationContext().containsBean(errorChannelName)) {
@@ -670,7 +680,7 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
BridgeHandler errorBridge = new BridgeHandler();
errorBridge.setOutputChannel(defaultErrorChannel);
errorChannel.subscribe(errorBridge);
String errorBridgeHandlerName = getErrorBridgeName(destination);
String errorBridgeHandlerName = getErrorBridgeName(destination, bindingName);
((GenericApplicationContext) getApplicationContext()).registerBean(
errorBridgeHandlerName, BridgeHandler.class, () -> errorBridge);
}
@@ -696,17 +706,19 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
if (!StringUtils.hasText(bindingName)) {
return;
}
BindingServiceProperties bsp = getApplicationContext().getBean(BindingServiceProperties.class);
BindingProperties bp = bsp.getBindingProperties(bindingName);
if (StringUtils.hasText(bp.getErrorHandlerDefinition())) {
FunctionCatalog catalog = getApplicationContext().getBean(FunctionCatalog.class);
Consumer<ErrorMessage> errorHandler = catalog.lookup(Consumer.class, bp.getErrorHandlerDefinition());
if (errorHandler == null) {
logger.warn("Failed to retrieve error handling function with definition: " + bp.getErrorHandlerDefinition() + ", for binding: " + bindingName);
}
else {
SubscribableChannel functionErrorChannel = getApplicationContext().getBean(errorChannelName, SubscribableChannel.class);
functionErrorChannel.subscribe(errorMessage -> errorHandler.accept((ErrorMessage) errorMessage));
BindingServiceProperties bsp = this.getBindingServiceProperties();
if (bsp != null) {
BindingProperties bp = bsp.getBindingProperties(bindingName);
if (bp != null && StringUtils.hasText(bp.getErrorHandlerDefinition())) {
FunctionCatalog catalog = getApplicationContext().getBean(FunctionCatalog.class);
Consumer<ErrorMessage> errorHandler = catalog.lookup(Consumer.class, bp.getErrorHandlerDefinition());
if (errorHandler == null) {
logger.warn("Failed to retrieve error handling function with definition: " + bp.getErrorHandlerDefinition() + ", for binding: " + bindingName);
}
else {
SubscribableChannel functionErrorChannel = getApplicationContext().getBean(errorChannelName, SubscribableChannel.class);
functionErrorChannel.subscribe(errorMessage -> errorHandler.accept((ErrorMessage) errorMessage));
}
}
}
}
@@ -841,9 +853,9 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
: true;
}
private void destroyErrorInfrastructure(ProducerDestination destination) {
String errorChannelName = errorsBaseName(destination);
String errorBridgeHandlerName = getErrorBridgeName(destination);
private void destroyErrorInfrastructure(ProducerDestination destination, String bindingName) {
String errorChannelName = errorsBaseName(destination, bindingName);
String errorBridgeHandlerName = getErrorBridgeName(destination, bindingName);
MessageHandler bridgeHandler = null;
if (getApplicationContext().containsBean(errorBridgeHandlerName)) {
bridgeHandler = getApplicationContext().getBean(errorBridgeHandlerName,
@@ -980,15 +992,19 @@ public abstract class AbstractMessageChannelBinder<C extends ConsumerProperties,
protected String errorsBaseName(ConsumerDestination destination, String group,
C consumerProperties) {
return destination.getName() + "." + group + ".errors";
return this.doErrorBaseName(consumerProperties.getBindingName());
}
protected String getErrorBridgeName(ProducerDestination destination) {
return errorsBaseName(destination) + ".bridge" + destination.hashCode();
protected String getErrorBridgeName(ProducerDestination destination, String bindingName) {
return errorsBaseName(destination, bindingName) + ".bridge" + destination.hashCode();
}
protected String errorsBaseName(ProducerDestination destination) {
return destination.getName() + ".errors";
protected String errorsBaseName(ProducerDestination destination, String bindingName) {
return this.doErrorBaseName(bindingName);
}
private String doErrorBaseName(String bindingName) {
return this.getBinderIdentity() + "." + bindingName + ".errors";
}
private Map<String, Object> doGetExtendedInfo(Object destination, Object properties) {

10
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/Binder.java
View File

@@ -30,10 +30,20 @@ package org.springframework.cloud.stream.binder;
* @author Jennifer Hickey
* @author Ilayaperumal Gopinathan
* @author Marius Bogoevici
* @author Oleg Zhurakousky
* @since 1.0
*/
public interface Binder<T, C extends ConsumerProperties, P extends ProducerProperties> {
/**
* Returns instance identity of this binder.
* Individual binders should normally override this method.
* @return instance identity of this binder
*/
default String getBinderIdentity() {
return String.valueOf(this.hashCode());
}
/**
* Bind the target component as a message consumer to the logical entity identified by
* the name.

6
core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/test/TestChannelBinderTests.java
View File

@@ -43,7 +43,7 @@ public class TestChannelBinderTests {
"--spring.cloud.stream.bindings.function-in-0.destination=input")) {
TestChannelBinder binder = context.getBean(TestChannelBinder.class);
Method registerErrorInfrastructure = ReflectionUtils
.findMethod(TestChannelBinder.class, "registerErrorInfrastructure", ProducerDestination.class);
.findMethod(TestChannelBinder.class, "registerErrorInfrastructure", ProducerDestination.class, String.class);
registerErrorInfrastructure.setAccessible(true);
ProducerDestination destination = new ProducerDestination() {
@Override
@@ -56,7 +56,7 @@ public class TestChannelBinderTests {
return "sample";
}
};
registerErrorInfrastructure.invoke(binder, destination);
registerErrorInfrastructure.invoke(binder, destination, "function-in-0");
destination = new ProducerDestination() {
@Override
public String getNameForPartition(int partition) {
@@ -68,7 +68,7 @@ public class TestChannelBinderTests {
return "sample";
}
};
registerErrorInfrastructure.invoke(binder, destination);
registerErrorInfrastructure.invoke(binder, destination, "function-in-0");
}
}