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

GH-1459: Pollable Consumer and Requeue

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream/issues/1459

Requires https://github.com/spring-cloud/spring-cloud-stream/pull/1467
This commit is contained in:
Gary Russell
2018-09-06 12:59:57 -04:00
parent e869516e3d
commit 36361d19bf
1 changed files with 34 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -35,6 +35,7 @@ import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.admin.CreateTopicsResult;
@@ -75,6 +76,7 @@ import org.springframework.cloud.stream.binder.HeaderMode;
import org.springframework.cloud.stream.binder.PartitionCapableBinderTests;
import org.springframework.cloud.stream.binder.PartitionTestSupport;
import org.springframework.cloud.stream.binder.PollableSource;
import org.springframework.cloud.stream.binder.RequeueCurrentMessageException;
import org.springframework.cloud.stream.binder.Spy;
import org.springframework.cloud.stream.binder.TestUtils;
import org.springframework.cloud.stream.binder.kafka.properties.KafkaBinderConfigurationProperties;
@@ -2533,6 +2535,38 @@ public class KafkaBinderTests extends
binding.unbind();
}
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void testPolledConsumerRequeue() throws Exception {
KafkaTestBinder binder = getBinder();
PollableSource<MessageHandler> inboundBindTarget = new DefaultPollableMessageSource(this.messageConverter);
ExtendedConsumerProperties<KafkaConsumerProperties> properties = createConsumerProperties();
Binding<PollableSource<MessageHandler>> binding = binder.bindPollableConsumer("pollableRequeue", "group",
inboundBindTarget, properties);
Map<String, Object> producerProps = KafkaTestUtils.producerProps(embeddedKafka.getEmbeddedKafka());
KafkaTemplate template = new KafkaTemplate(new DefaultKafkaProducerFactory<>(producerProps));
template.send("pollableRequeue", "testPollable");
try {
boolean polled = false;
int n = 0;
while (n++ < 100 && !polled) {
polled = inboundBindTarget.poll(m -> {
assertThat(m.getPayload()).isEqualTo("testPollable".getBytes());
throw new RequeueCurrentMessageException();
});
}
fail("Expected exception");
}
catch (MessageHandlingException e) {
assertThat(e.getCause()).isInstanceOf(RequeueCurrentMessageException.class);
}
boolean polled = inboundBindTarget.poll(m -> {
assertThat(m.getPayload()).isEqualTo("testPollable".getBytes());
});
assertThat(polled).isTrue();
binding.unbind();
}
@SuppressWarnings({ "rawtypes", "unchecked" })
@Test
public void testPolledConsumerWithDlq() throws Exception {