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

GH-313: Fix DLQ Transactions and Confirms/Returns

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit/issues/313

- Detect a transactional consumer and publish to the DLQ in the same transaction.
- Detect confirms/returns config and wait for confirm if so configured.
- Add tests for transactional publication, both types of confirms.
- Returned message and negative ack tested manually.

Resolves #314
This commit is contained in:
Gary Russell
2021-02-09 14:17:55 -05:00
committed by Oleg Zhurakousky
parent 8b6a0713ff
commit db0af5e96f
3 changed files with 213 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
docs/src/main/asciidoc/overview.adoc
View File

@@ -57,6 +57,11 @@ Starting with version 2.1, this is true regardless of the setting of `republishT
IMPORTANT: Setting `requeueRejected` to `true` (with `republishToDlq=false` ) causes the message to be re-queued and redelivered continually, which is likely not what you want unless the reason for the failure is transient.
In general, you should enable retry within the binder by setting `maxAttempts` to greater than one or by setting `republishToDlq` to `true`.
Starting with version 3.1.2, if the consumer is marked as `transacted`, publishing to the DLQ will participate in the transaction.
This allows the transaction to roll back if the publishing fails for some reason (for example, if the user is not authorized to publish to the dead letter exchange).
In addition, if the connection factory is configured for publisher confirms or returns, the publication to the DLQ will wait for the confirmation and check for a returned message.
If a negative acknowledgment or returned message is received, the binder will throw an `AmqpRejectAndDontRequeueException`, allowing the broker to take care of publishing to the DLQ as if the `republishToDlq` property is `false`.
See <<rabbit-binder-properties>> for more information about these properties.
The framework does not provide any standard mechanism to consume dead-letter messages (or to re-route them back to the primary queue).

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

@@ -23,6 +23,9 @@ import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;
@@ -39,7 +42,10 @@ import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.batch.BatchingStrategy;
import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory.ConfirmType;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.connection.CorrelationData.Confirm;
import org.springframework.amqp.rabbit.connection.LocalizedQueueConnectionFactory;
import org.springframework.amqp.rabbit.connection.RabbitUtils;
import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
@@ -618,11 +624,27 @@ public class RabbitMessageChannelBinder extends
if (properties.getExtension().isRepublishToDlq()) {
return new MessageHandler() {
private static final long ACK_TIMEOUT = 10_000;
private final RabbitTemplate template = new RabbitTemplate(
RabbitMessageChannelBinder.this.connectionFactory);
private final ConfirmType confirmType;
{
this.template.setUsePublisherConnection(true);
this.template.setChannelTransacted(properties.getExtension().isTransacted());
this.template.setMandatory(RabbitMessageChannelBinder.this.connectionFactory.isPublisherReturns());
if (RabbitMessageChannelBinder.this.connectionFactory.isSimplePublisherConfirms()) {
this.confirmType = ConfirmType.SIMPLE;
}
else if (RabbitMessageChannelBinder.this.connectionFactory.isPublisherConfirms()) {
this.confirmType = ConfirmType.CORRELATED;
}
else {
this.confirmType = ConfirmType.NONE;
}
}
private final String exchange = deadLetterExchangeName(properties.getExtension());
@@ -694,7 +716,7 @@ public class RabbitMessageChannelBinder extends
messageProperties.setDeliveryMode(
properties.getExtension().getRepublishDeliveyMode());
}
this.template.send(this.exchange,
doSend(this.exchange,
this.routingKey != null ? this.routingKey
: messageProperties.getConsumerQueue(),
amqpMessage);
@@ -716,6 +738,45 @@ public class RabbitMessageChannelBinder extends
}
}
private void doSend(String exchange, String routingKey, Message amqpMessage) {
if (ConfirmType.SIMPLE.equals(this.confirmType)) {
this.template.invoke(temp -> {
temp.send(exchange, routingKey, amqpMessage);
if (!temp.waitForConfirms(ACK_TIMEOUT)) {
throw new AmqpRejectAndDontRequeueException("Negative ack for DLQ message received");
}
return null;
});
}
else if (ConfirmType.CORRELATED.equals(this.confirmType)) {
CorrelationData corr = new CorrelationData();
this.template.send(exchange, routingKey, amqpMessage, corr);
try {
Confirm confirm = corr.getFuture().get(ACK_TIMEOUT, TimeUnit.MILLISECONDS);
if (!confirm.isAck()) {
throw new AmqpRejectAndDontRequeueException("Negative ack for DLQ message received");
}
}
catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new AmqpRejectAndDontRequeueException(e);
}
catch (ExecutionException e) {
throw new AmqpRejectAndDontRequeueException(e.getCause());
}
catch (TimeoutException e) {
throw new AmqpRejectAndDontRequeueException(e);
}
if (corr.getReturned() != null) {
RabbitMessageChannelBinder.this.logger.error("DLQ message was returned: " + amqpMessage);
throw new AmqpRejectAndDontRequeueException("DLQ message was returned");
}
}
else {
this.template.send(exchange, routingKey, amqpMessage);
}
}
private boolean checkDlx() {
if (this.dlxPresent == null) {
if (properties.getExtension().isAutoBindDlq()) {

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

@@ -29,11 +29,13 @@ import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.zip.Deflater;
import com.rabbitmq.client.LongString;
@@ -115,6 +117,7 @@ import org.springframework.integration.channel.DirectChannel;
import org.springframework.integration.channel.QueueChannel;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.integration.expression.ValueExpression;
import org.springframework.integration.handler.BridgeHandler;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageChannel;
@@ -167,10 +170,7 @@ public class RabbitBinderTests extends
protected RabbitTestBinder getBinder() {
if (this.testBinder == null) {
RabbitProperties rabbitProperties = new RabbitProperties();
rabbitProperties.setPublisherConfirmType(ConfirmType.SIMPLE);
rabbitProperties.setPublisherReturns(true);
this.testBinder = new RabbitTestBinder(rabbitAvailableRule.getResource(),
rabbitProperties);
this.testBinder = new RabbitTestBinder(this.rabbitAvailableRule.getResource(), rabbitProperties);
}
return this.testBinder;
}
@@ -1520,6 +1520,148 @@ public class RabbitBinderTests extends
consumerBinding.unbind();
}
@SuppressWarnings("unchecked")
@Test
public void testAutoBindDLQwithRepublishTx() throws Exception {
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
consumerProperties.getExtension().setPrefix(TEST_PREFIX);
consumerProperties.getExtension().setAutoBindDlq(true);
consumerProperties.getExtension().setRepublishToDlq(true);
consumerProperties.setMaxAttempts(1); // disable retry
consumerProperties.getExtension().setDurableSubscription(true);
consumerProperties.getExtension().setTransacted(true);
DirectChannel moduleInputChannel = createBindableChannel("input",
createConsumerBindingProperties(consumerProperties));
moduleInputChannel.setBeanName("dlqPubTestTx");
moduleInputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("test");
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer(
"foo.dlqpubtestTx", "foo", moduleInputChannel, consumerProperties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtestTx.foo", "foo");
template.setReceiveTimeout(10_000);
org.springframework.amqp.core.Message deadLetter = template
.receive(TEST_PREFIX + "foo.dlqpubtestTx.foo.dlq");
assertThat(deadLetter).isNotNull();
assertThat(deadLetter.getBody()).isEqualTo("foo".getBytes());
assertThat(deadLetter.getMessageProperties().getHeaders())
.containsKey((RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE));
List<Object> errorHandler = (List<Object>) TestUtils.getPropertyValue(consumerBinding,
"lifecycle.errorChannel.dispatcher.handlers", Set.class).stream()
.filter(handler -> !handler.getClass().equals(BridgeHandler.class))
.collect(Collectors.toList());
assertThat(errorHandler).hasSize(1);
assertThat(TestUtils.getPropertyValue(errorHandler.get(0), "template.transactional", Boolean.class)).isTrue();
assertThat(TestUtils.getPropertyValue(errorHandler.get(0), "confirmType", ConfirmType.class))
.isEqualTo(ConfirmType.NONE);
consumerBinding.unbind();
}
@SuppressWarnings("unchecked")
@Test
public void testAutoBindDLQwithRepublishSimpleConfirms() throws Exception {
CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
ccf.setPublisherReturns(true);
ccf.setPublisherConfirmType(ConfirmType.SIMPLE);
ccf.resetConnection();
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
consumerProperties.getExtension().setPrefix(TEST_PREFIX);
consumerProperties.getExtension().setAutoBindDlq(true);
consumerProperties.getExtension().setRepublishToDlq(true);
consumerProperties.setMaxAttempts(1); // disable retry
consumerProperties.getExtension().setDurableSubscription(true);
DirectChannel moduleInputChannel = createBindableChannel("input",
createConsumerBindingProperties(consumerProperties));
moduleInputChannel.setBeanName("dlqPubtestSimple");
moduleInputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("test");
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer(
"foo.dlqpubtestSimple", "foo", moduleInputChannel, consumerProperties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtestSimple.foo", "foo");
template.setReceiveTimeout(10_000);
org.springframework.amqp.core.Message deadLetter = template
.receive(TEST_PREFIX + "foo.dlqpubtestSimple.foo.dlq");
assertThat(deadLetter).isNotNull();
assertThat(deadLetter.getBody()).isEqualTo("foo".getBytes());
assertThat(deadLetter.getMessageProperties().getHeaders())
.containsKey((RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE));
List<Object> errorHandler = (List<Object>) TestUtils.getPropertyValue(consumerBinding,
"lifecycle.errorChannel.dispatcher.handlers", Set.class).stream()
.filter(handler -> !handler.getClass().equals(BridgeHandler.class))
.collect(Collectors.toList());
assertThat(errorHandler).hasSize(1);
assertThat(TestUtils.getPropertyValue(errorHandler.get(0), "confirmType", ConfirmType.class))
.isEqualTo(ConfirmType.SIMPLE);
consumerBinding.unbind();
}
@SuppressWarnings("unchecked")
@Test
public void testAutoBindDLQwithRepublishCorrelatedConfirms() throws Exception {
CachingConnectionFactory ccf = this.rabbitAvailableRule.getResource();
ccf.setPublisherReturns(true);
ccf.setPublisherConfirmType(ConfirmType.CORRELATED);
ccf.resetConnection();
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> consumerProperties = createConsumerProperties();
consumerProperties.getExtension().setPrefix(TEST_PREFIX);
consumerProperties.getExtension().setAutoBindDlq(true);
consumerProperties.getExtension().setRepublishToDlq(true);
consumerProperties.setMaxAttempts(1); // disable retry
consumerProperties.getExtension().setDurableSubscription(true);
DirectChannel moduleInputChannel = createBindableChannel("input",
createConsumerBindingProperties(consumerProperties));
moduleInputChannel.setBeanName("dlqPubtestCorrelated");
moduleInputChannel.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
throw new RuntimeException("test");
}
});
Binding<MessageChannel> consumerBinding = binder.bindConsumer(
"foo.dlqpubtestCorrelated", "foo", moduleInputChannel, consumerProperties);
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.convertAndSend("", TEST_PREFIX + "foo.dlqpubtestCorrelated.foo", "foo");
template.setReceiveTimeout(10_000);
org.springframework.amqp.core.Message deadLetter = template
.receive(TEST_PREFIX + "foo.dlqpubtestCorrelated.foo.dlq");
assertThat(deadLetter).isNotNull();
assertThat(deadLetter.getBody()).isEqualTo("foo".getBytes());
assertThat(deadLetter.getMessageProperties().getHeaders())
.containsKey((RepublishMessageRecoverer.X_EXCEPTION_STACKTRACE));
List<Object> errorHandler = (List<Object>) TestUtils.getPropertyValue(consumerBinding,
"lifecycle.errorChannel.dispatcher.handlers", Set.class).stream()
.filter(handler -> !handler.getClass().equals(BridgeHandler.class))
.collect(Collectors.toList());
assertThat(errorHandler).hasSize(1);
assertThat(TestUtils.getPropertyValue(errorHandler.get(0), "confirmType", ConfirmType.class))
.isEqualTo(ConfirmType.CORRELATED);
consumerBinding.unbind();
}
@SuppressWarnings("unchecked")
@Test
public void testBatchingAndCompression() throws Exception {