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

GH-13: Fix republishToDlq with Partitions

Browse Source 
Fixes #13
This commit is contained in:
Gary Russell
2016-09-16 14:51:51 -04:00
parent 1ce8c555f2
commit d8a861f750
2 changed files with 103 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -20,8 +20,6 @@ import java.util.Arrays;
import java.util.HashMap;
import java.util.Map;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Envelope;
import org.aopalliance.aop.Advice;
import org.springframework.amqp.AmqpConnectException;
@@ -71,6 +69,9 @@ import org.springframework.scheduling.TaskScheduler;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Envelope;
/**
* A {@link org.springframework.cloud.stream.binder.Binder} implementation backed by RabbitMQ.
* @author Mark Fisher
@@ -220,7 +221,7 @@ public class RabbitMessageChannelBinder
listenerContainer.setRecoveryInterval(properties.getExtension().getRecoveryInterval());
listenerContainer.setTxSize(properties.getExtension().getTxSize());
listenerContainer.setTaskExecutor(new SimpleAsyncTaskExecutor(destination.getName() + "-"));
listenerContainer.setQueues((Queue) destination);
listenerContainer.setQueues(destination);
if (properties.getMaxAttempts() > 1 || properties.getExtension().isRepublishToDlq()) {
RetryOperationsInterceptor retryInterceptor = RetryInterceptorBuilder.stateless()
.retryOperations(buildRetryTemplate(properties))
@@ -417,23 +418,25 @@ public class RabbitMessageChannelBinder
/**
* If so requested, declare the DLX/DLQ and bind it. The DLQ is bound to the DLX with a routing key of the original
* queue name because we use default exchange routing by queue name for the original message.
* @param queueName The base name for the queue (including the binder prefix, if any).
* @param baseQueueName The base name for the queue (including the binder prefix, if any).
* @param routingKey The routing key for the queue.
* @param autoBindDlq true if the DLQ should be bound.
*/
private void autoBindDLQ(final String queueName, String routingKey, String prefix, boolean autoBindDlq) {
private void autoBindDLQ(final String baseQueueName, String routingKey, String prefix, boolean autoBindDlq) {
if (this.logger.isDebugEnabled()) {
this.logger.debug("autoBindDLQ=" + autoBindDlq
+ " for: " + queueName);
+ " for: " + baseQueueName);
}
if (autoBindDlq) {
String dlqName = constructDLQName(queueName);
String dlqName = constructDLQName(baseQueueName);
Queue dlq = new Queue(dlqName);
declareQueue(dlqName, dlq);
final String dlxName = deadLetterExchangeName(prefix);
final DirectExchange dlx = new DirectExchange(dlxName);
declareExchange(dlxName, dlx);
declareBinding(dlqName, BindingBuilder.bind(dlq).to(dlx).with(routingKey));
// Also bind with the base queue name in case republishToDlq is used, which does not know about partitioning
declareBinding(dlqName, BindingBuilder.bind(dlq).to(dlx).with(baseQueueName));
}
}

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

@@ -16,6 +16,11 @@
package org.springframework.cloud.stream.binder.rabbit;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -64,11 +69,6 @@ import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.GenericMessage;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.spy;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
/**
* @author Mark Fisher
* @author Gary Russell
@@ -438,6 +438,94 @@ public class RabbitBinderTests extends
outputBinding.unbind();
}
@Test
public void testAutoBindDLQPartionedConsumerFirstWithRepublish() throws Exception {
RabbitTestBinder binder = getBinder();
ExtendedConsumerProperties<RabbitConsumerProperties> properties = createConsumerProperties();
properties.getExtension().setPrefix("bindertest.");
properties.getExtension().setAutoBindDlq(true);
properties.getExtension().setRepublishToDlq(true);
properties.setMaxAttempts(1); // disable retry
properties.setPartitioned(true);
properties.setInstanceIndex(0);
DirectChannel input0 = createBindableChannel("input", createConsumerBindingProperties(properties));
input0.setBeanName("test.input0DLQ");
Binding<MessageChannel> input0Binding = binder.bindConsumer("partPubDLQ.0", "dlqPartGrp", input0, properties);
Binding<MessageChannel> defaultConsumerBinding1 = binder.bindConsumer("partPubDLQ.0", "default",
new QueueChannel(), properties);
properties.setInstanceIndex(1);
DirectChannel input1 = createBindableChannel("input1", createConsumerBindingProperties(properties));
input1.setBeanName("test.input1DLQ");
Binding<MessageChannel> input1Binding = binder.bindConsumer("partPubDLQ.0", "dlqPartGrp", input1, properties);
Binding<MessageChannel> defaultConsumerBinding2 = binder.bindConsumer("partPubDLQ.0", "default",
new QueueChannel(), properties);
ExtendedProducerProperties<RabbitProducerProperties> producerProperties = createProducerProperties();
producerProperties.getExtension().setPrefix("bindertest.");
producerProperties.getExtension().setAutoBindDlq(true);
producerProperties.setPartitionKeyExtractorClass(PartitionTestSupport.class);
producerProperties.setPartitionSelectorClass(PartitionTestSupport.class);
producerProperties.setPartitionCount(2);
BindingProperties bindingProperties = createProducerBindingProperties(producerProperties);
DirectChannel output = createBindableChannel("output", bindingProperties);
output.setBeanName("test.output");
Binding<MessageChannel> outputBinding = binder.bindProducer("partPubDLQ.0", output, producerProperties);
final CountDownLatch latch0 = new CountDownLatch(1);
input0.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (latch0.getCount() <= 0) {
throw new RuntimeException("dlq");
}
latch0.countDown();
}
});
final CountDownLatch latch1 = new CountDownLatch(1);
input1.subscribe(new MessageHandler() {
@Override
public void handleMessage(Message<?> message) throws MessagingException {
if (latch1.getCount() <= 0) {
throw new RuntimeException("dlq");
}
latch1.countDown();
}
});
output.send(new GenericMessage<>(1));
assertThat(latch1.await(10, TimeUnit.SECONDS)).isTrue();
output.send(new GenericMessage<>(0));
assertThat(latch0.await(10, TimeUnit.SECONDS)).isTrue();
output.send(new GenericMessage<>(1));
RabbitTemplate template = new RabbitTemplate(this.rabbitAvailableRule.getResource());
template.setReceiveTimeout(10000);
String streamDLQName = "bindertest.partPubDLQ.0.dlqPartGrp.dlq";
org.springframework.amqp.core.Message received = template.receive(streamDLQName);
assertThat(received).isNotNull();
assertThat(received.getMessageProperties().getHeaders()).containsEntry("partition", 1);
output.send(new GenericMessage<>(0));
received = template.receive(streamDLQName);
assertThat(received).isNotNull();
assertThat(received.getMessageProperties().getHeaders()).containsEntry("partition", 0);
input0Binding.unbind();
input1Binding.unbind();
defaultConsumerBinding1.unbind();
defaultConsumerBinding2.unbind();
outputBinding.unbind();
}
@Test
public void testAutoBindDLQPartitionedProducerFirst() throws Exception {
RabbitTestBinder binder = getBinder();