From 3de0445aaab99ad96c655918851e0974e6457ce9 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 7 Aug 2020 15:26:57 -0400 Subject: [PATCH] GH-3358: Kafka IB Gateway - implement Pausable Resolves https://github.com/spring-projects/spring-integration/issues/3358 Also add `isPaused()` to `Pausable`. * Fix `@since` to the proper version --- .../integration/core/Pausable.java | 11 ++++++++++ .../kafka/inbound/KafkaInboundGateway.java | 18 ++++++++++++++- .../KafkaMessageDrivenChannelAdapter.java | 5 +++++ .../kafka/inbound/KafkaMessageSource.java | 22 +++++-------------- .../kafka/inbound/InboundGatewayTests.java | 21 +++++++++++++++++- .../inbound/MessageDrivenAdapterTests.java | 2 ++ 6 files changed, 60 insertions(+), 19 deletions(-) diff --git a/spring-integration-core/src/main/java/org/springframework/integration/core/Pausable.java b/spring-integration-core/src/main/java/org/springframework/integration/core/Pausable.java index 4fb4fb1325..cde8fd6ee7 100644 --- a/spring-integration-core/src/main/java/org/springframework/integration/core/Pausable.java +++ b/spring-integration-core/src/main/java/org/springframework/integration/core/Pausable.java @@ -17,6 +17,7 @@ package org.springframework.integration.core; import org.springframework.integration.support.management.ManageableLifecycle; +import org.springframework.jmx.export.annotation.ManagedAttribute; import org.springframework.jmx.export.annotation.ManagedOperation; /** @@ -42,4 +43,14 @@ public interface Pausable extends ManageableLifecycle { @ManagedOperation(description = "Resume the component") void resume(); + /** + * Check if the endpoint is paused. + * @return true if paused. + * @since 5.4 + */ + @ManagedAttribute(description = "Is the component paused?") + default boolean isPaused() { + throw new UnsupportedOperationException("This component does not implement this method"); + } + } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java index d301299cc3..f58c557b28 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java @@ -29,6 +29,7 @@ import org.apache.kafka.common.header.Header; import org.springframework.core.AttributeAccessor; import org.springframework.integration.IntegrationMessageHeaderAccessor; import org.springframework.integration.context.OrderlyShutdownCapable; +import org.springframework.integration.core.Pausable; import org.springframework.integration.gateway.MessagingGatewaySupport; import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; @@ -68,7 +69,7 @@ import org.springframework.util.Assert; * @since 5.4 * */ -public class KafkaInboundGateway extends MessagingGatewaySupport implements OrderlyShutdownCapable { +public class KafkaInboundGateway extends MessagingGatewaySupport implements Pausable, OrderlyShutdownCapable { private static final ThreadLocal ATTRIBUTES_HOLDER = new ThreadLocal<>(); @@ -194,6 +195,21 @@ public class KafkaInboundGateway extends MessagingGatewaySupport implem this.messageListenerContainer.stop(); } + @Override + public void pause() { + this.messageListenerContainer.pause(); + } + + @Override + public void resume() { + this.messageListenerContainer.resume(); + } + + @Override + public boolean isPaused() { + return this.messageListenerContainer.isContainerPaused(); + } + @Override public String getComponentType() { return "kafka:inbound-gateway"; diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java index a59a6d8856..4ab767132c 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java @@ -339,6 +339,11 @@ public class KafkaMessageDrivenChannelAdapter extends MessageProducerSuppo this.messageListenerContainer.resume(); } + @Override + public boolean isPaused() { + return this.messageListenerContainer.isContainerPaused(); + } + @Override public int beforeShutdown() { this.messageListenerContainer.stop(); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java index 4a13f09d25..12dc062aa5 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java @@ -246,15 +246,6 @@ public class KafkaMessageSource extends AbstractMessageSource impl return Collections.unmodifiableCollection(this.assignedPartitions); } - /** - * Return true if the source is currently paused. - * @return true if paused. - * @since 3.2.2 - */ - public boolean isPaused() { - return this.paused; - } - @Override protected void onInit() { if (!StringUtils.hasText(this.consumerProperties.getClientId())) { @@ -406,24 +397,21 @@ public class KafkaMessageSource extends AbstractMessageSource impl this.running = false; } - /** - * {@inheritDoc} - * @since 3.1.2 - */ @Override public synchronized void pause() { this.pausing = true; } - /** - * {@inheritDoc} - * @since 3.1.2 - */ @Override public synchronized void resume() { this.pausing = false; } + @Override + public boolean isPaused() { + return this.paused; + } + @Override protected synchronized Object doReceive() { if (this.consumer == null) { diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java index 1702ad0f51..68adf82bc8 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2018-2020 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. @@ -45,6 +45,8 @@ 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.event.ConsumerPausedEvent; +import org.springframework.kafka.event.ConsumerResumedEvent; import org.springframework.kafka.listener.ContainerProperties; import org.springframework.kafka.listener.KafkaMessageListenerContainer; import org.springframework.kafka.support.Acknowledgment; @@ -118,8 +120,19 @@ class InboundGatewayTests { DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties(topic1); containerProps.setIdleEventInterval(100L); + containerProps.setPollTimeout(200L); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); + CountDownLatch pausedLatch = new CountDownLatch(1); + CountDownLatch resumedLatch = new CountDownLatch(1); + container.setApplicationEventPublisher(event -> { + if (event instanceof ConsumerPausedEvent) { + pausedLatch.countDown(); + } + else if (event instanceof ConsumerResumedEvent) { + resumedLatch.countDown(); + } + }); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); ProducerFactory pf = new DefaultKafkaProducerFactory<>(senderProps); KafkaTemplate template = new KafkaTemplate<>(pf); @@ -175,6 +188,12 @@ class InboundGatewayTests { assertThat(record).has(partition(1)); assertThat(record).has(value("FOO")); assertThat(onPartitionsAssignedCalledLatch.await(10, TimeUnit.SECONDS)).isTrue(); + gateway.pause(); + assertThat(pausedLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(gateway.isPaused()).isTrue(); + gateway.resume(); + assertThat(resumedLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(gateway.isPaused()).isFalse(); gateway.stop(); } diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java index 953dd27ba4..f271832948 100644 --- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java +++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java @@ -607,8 +607,10 @@ class MessageDrivenAdapterTests { assertThat(outputChannel.getQueueSize()).isEqualTo(2); adapter.pause(); assertThat(pauseLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(adapter.isPaused()).isTrue(); adapter.resume(); assertThat(resumeLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(adapter.isPaused()).isFalse(); adapter.stop(); }