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

GH-2921: Output-bindings and RabbitMQ r-k-e issues

Browse Source 
* When output-bindings config is explicitly used for StreamBridge
  and the RabbitMQ routing-key-expression is provided, Spring Cloud Stream
  is throwing an exception due to a proper function is not found in
  the catalog. Bypassing this step and letting the bootstrapping continues
  if output-bindings and RabbitMQ routing-key-expression combination is used.
* Adding tests to verify

Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/2921
This commit is contained in:
Soby Chacko
2024-03-20 16:14:16 -04:00
committed by GitHub
parent f836e1cf54
commit 9190e9b835
3 changed files with 56 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -217,6 +217,21 @@ class StreamBridgeTests {
}
}
// See for more details: https://github.com/spring-cloud/spring-cloud-stream/issues/2921
@Test
void outputBindingsAndRabbitRoutingKeyExpressionComboDoesNotCauseIssues() {
try (ConfigurableApplicationContext context = new SpringApplicationBuilder(
TestChannelBinderConfiguration.getCompleteConfiguration(
EmptyConfiguration.class)).web(WebApplicationType.NONE).run(
"--spring.cloud.stream.output-bindings=the-exchange",
"--spring.cloud.stream.rabbit.bindings.the-exchange.producer.routing-key-expression=headers['key']")) {
// Ensure no exception is thrown from the above properties.
// See this issue for more details: https://github.com/spring-cloud/spring-cloud-stream/issues/2921
// Without the corresponding chages added in this PR, this config would throw an exception.
}
}
// Fore more context on this test: https://github.com/spring-cloud/spring-cloud-stream/issues/2848
@Test
void ensurePartitioningWorksWhenNativeEncodingEnabledAndOutputBindingsExist() {

39
core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binding/MessageConverterConfigurer.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2015-2023 the original author or authors.
* Copyright 2015-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -167,7 +167,24 @@ public class MessageConverterConfigurer
if (environment != null && environment.containsProperty("spring.cloud.stream.rabbit.bindings." + channelName + ".producer.routing-key-expression")) {
Map<String, String> channelNameToFunctions = BindableFunctionProxyFactory.getChannelNameToFunctions();
String functionName = channelNameToFunctions.get(channelName);
this.setSkipOutputConversionIfNecessary(functionName);
String outputBindings = environment.getProperty("spring.cloud.stream.output-bindings");
FunctionInvocationWrapper function = null;
if (outputBindings != null && outputBindings.contains(functionName)) {
try {
function = retrieveFunction(functionName);
}
catch (Exception e) {
// we expect an exception in this case since we didn't have a matching function
// for the function name due to output-bindings, hence passing through.
// See https://github.com/spring-cloud/spring-cloud-stream/issues/2921 for more details.
}
}
else {
function = retrieveFunction(functionName);
}
if (function != null) {
function.setSkipOutputConversion(true);
}
functional = false;
}
if (!functional) {
@@ -177,13 +194,19 @@ public class MessageConverterConfigurer
}
}
private void setSkipOutputConversionIfNecessary(String functionName) {
private FunctionInvocationWrapper retrieveFunction(String functionName) {
FunctionCatalog catalog = this.beanFactory.getBean(FunctionCatalog.class);
if (catalog != null) {
FunctionInvocationWrapper function = catalog.lookup(functionName);
if (function != null) {
function.setSkipOutputConversion(true);
}
FunctionInvocationWrapper function = catalog.lookup(functionName);
if (function != null) {
function.setSkipOutputConversion(true);
}
return function;
}
private void skipOutputConversionIfNecessary(String functionName) {
FunctionInvocationWrapper function = retrieveFunction(functionName);
if (function != null) {
function.setSkipOutputConversion(true);
}
}

31
core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binding/MessageConverterConfigurerTests.java
View File

@@ -1,5 +1,5 @@
/*
* Copyright 2017-2023 the original author or authors.
* Copyright 2017-2024 the original author or authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
@@ -25,26 +25,25 @@ import org.springframework.cloud.stream.config.BindingProperties;
import org.springframework.cloud.stream.config.BindingServiceProperties;
import org.springframework.cloud.stream.converter.CompositeMessageConverterFactory;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageDeliveryException;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.converter.AbstractMessageConverter;
import org.springframework.messaging.converter.MessageConversionException;
import org.springframework.messaging.converter.MessageConverter;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.MimeType;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.fail;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
/**
* @author Gary Russell
* @author Oleg Zhurakousky
* @author Soby Chacko
* @since 1.3
*
*/
class MessageConverterConfigurerTests {
// @Test
@Test
void configureOutputChannelWithBadContentType() {
BindingServiceProperties props = new BindingServiceProperties();
BindingProperties bindingProps = new BindingProperties();
@@ -56,11 +55,8 @@ class MessageConverterConfigurerTests {
converterFactory.getMessageConverterForAllRegistered());
QueueChannel out = new QueueChannel();
configurer.configureOutputChannel(out, "foo");
out.send(new GenericMessage<Foo>(new Foo(), Collections
.<String, Object>singletonMap(MessageHeaders.CONTENT_TYPE, "bad/ct")));
Message<?> received = out.receive(0);
assertThat(received).isNotNull();
assertThat(received.getPayload()).isInstanceOf(Foo.class);
assertThatThrownBy(() -> out.send(new GenericMessage<>(new Foo(), Collections
.singletonMap(MessageHeaders.CONTENT_TYPE, "bad/ct")))).isInstanceOf(MessageDeliveryException.class);
}
@Test
@@ -91,16 +87,9 @@ class MessageConverterConfigurerTests {
converterFactory.getMessageConverterForAllRegistered());
QueueChannel out = new QueueChannel();
configurer.configureOutputChannel(out, "foo");
try {
out.send(new GenericMessage<Foo>(new Foo(),
Collections.<String, Object>singletonMap(MessageHeaders.CONTENT_TYPE,
"bad/ct")));
fail("Expected MessageConversionException: " + out.receive(0));
}
catch (MessageConversionException e) {
assertThat(e.getMessage())
.endsWith("to the configured output type: 'foo/bar'");
}
assertThatThrownBy(() -> out.send(new GenericMessage<>(new Foo(),
Collections.singletonMap(MessageHeaders.CONTENT_TYPE,
"bad/ct")))).isInstanceOf(MessageDeliveryException.class);
}
public static class Foo {