Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-198: Support pause/resume on inbound adapter

Browse Source 
Resolves https://github.com/spring-projects/spring-integration-kafka/issues/198

You can now pause/resume the consumer.

Any records previously fetched will be processed before the pause takes effect.
The listener container requires an `idleEventInterval`, a resume will take effect
on the next `ListenerContainerIdleEvent`.

* Delegate Pausable methods to listener container

Requires https://github.com/spring-projects/spring-kafka/pull/584

* Polishing
This commit is contained in:
Gary Russell
2018-02-22 08:48:58 -05:00
committed by Artem Bilan
parent 7b7bcf1014
commit 17ea7dace7
2 changed files with 92 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java
View File

@@ -26,6 +26,7 @@ import org.springframework.core.AttributeAccessor;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
import org.springframework.integration.context.OrderlyShutdownCapable;
import org.springframework.integration.endpoint.MessageProducerSupport;
import org.springframework.integration.endpoint.Pausable;
import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageStrategy;
import org.springframework.integration.support.ErrorMessageUtils;
@@ -66,7 +67,8 @@ import org.springframework.util.Assert;
* @author Artem Bilan
*
*/
public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSupport implements OrderlyShutdownCapable {
public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSupport implements OrderlyShutdownCapable,
Pausable {
private static final ThreadLocal<AttributeAccessor> attributesHolder = new ThreadLocal<>();
@@ -223,6 +225,11 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
this.batchListener.setFallbackType(payloadType);
}
@Override
public String getComponentType() {
return "kafka:message-driven-channel-adapter";
}
@Override
protected void onInit() {
super.onInit();
@@ -280,8 +287,13 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
}
@Override
public String getComponentType() {
return "kafka:message-driven-channel-adapter";
public void pause() {
this.messageListenerContainer.pause();
}
@Override
public void resume() {
this.messageListenerContainer.resume();
}
@Override
@@ -434,7 +446,8 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
@Override
public void onMessage(List<ConsumerRecord<K, V>> records, Acknowledgment acknowledgment,
Consumer<?, ?> consumer) {
Message<?> message = null;
Message<?> message = null;
try {
message = toMessagingMessage(records, acknowledgment, consumer);
setAttributesIfNecessary(records, message);

75
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
View File

@@ -17,17 +17,32 @@
package org.springframework.integration.kafka.inbound;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
import static org.mockito.BDDMockito.given;
import static org.mockito.BDDMockito.willAnswer;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import java.lang.reflect.Type;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.junit.ClassRule;
@@ -40,16 +55,19 @@ import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAd
import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.integration.support.StaticMessageHeaderAccessor;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.listener.config.ContainerProperties;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.KafkaNull;
import org.springframework.kafka.support.TopicPartitionInitialOffset;
import org.springframework.kafka.support.converter.BatchMessageConverter;
import org.springframework.kafka.support.converter.BatchMessagingMessageConverter;
import org.springframework.kafka.support.converter.ConversionException;
@@ -411,6 +429,63 @@ public class MessageDrivenAdapterTests {
adapter.stop();
}
@SuppressWarnings({ "unchecked", "rawtypes" })
@Test
public void testPauseResume() throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(isNull(), eq("clientId"), isNull())).willReturn(consumer);
final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = new HashMap<>();
records.put(new TopicPartition("foo", 0), Arrays.asList(
new ConsumerRecord<>("foo", 0, 0L, 1, "foo"),
new ConsumerRecord<>("foo", 0, 1L, 1, "bar")));
ConsumerRecords<Integer, String> consumerRecords = new ConsumerRecords<>(records);
ConsumerRecords<Integer, String> emptyRecords = new ConsumerRecords<>(Collections.emptyMap());
AtomicBoolean first = new AtomicBoolean(true);
given(consumer.poll(anyLong())).willAnswer(i -> {
Thread.sleep(50);
return first.getAndSet(false) ? consumerRecords : emptyRecords;
});
final CountDownLatch commitLatch = new CountDownLatch(2);
willAnswer(i -> {
commitLatch.countDown();
return null;
}).given(consumer).commitSync(any(Map.class));
given(consumer.assignment()).willReturn(records.keySet());
final CountDownLatch pauseLatch = new CountDownLatch(1);
willAnswer(i -> {
pauseLatch.countDown();
return null;
}).given(consumer).pause(records.keySet());
given(consumer.paused()).willReturn(records.keySet());
final CountDownLatch resumeLatch = new CountDownLatch(1);
willAnswer(i -> {
resumeLatch.countDown();
return null;
}).given(consumer).resume(records.keySet());
TopicPartitionInitialOffset[] topicPartition = new TopicPartitionInitialOffset[] {
new TopicPartitionInitialOffset("foo", 0) };
ContainerProperties containerProps = new ContainerProperties(topicPartition);
containerProps.setAckMode(AckMode.RECORD);
containerProps.setClientId("clientId");
containerProps.setIdleEventInterval(100L);
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
KafkaMessageDrivenChannelAdapter adapter = new KafkaMessageDrivenChannelAdapter(container);
QueueChannel outputChannel = new QueueChannel();
adapter.setOutputChannel(outputChannel);
adapter.afterPropertiesSet();
adapter.start();
assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue();
verify(consumer, times(2)).commitSync(any(Map.class));
assertThat(outputChannel.getQueueSize()).isEqualTo(2);
adapter.pause();
assertThat(pauseLatch.await(10, TimeUnit.SECONDS)).isTrue();
adapter.resume();
assertThat(resumeLatch.await(10, TimeUnit.SECONDS)).isTrue();
adapter.stop();
}
public static class Foo {
private String bar;