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

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");
+	}
+
 }

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<K, V, R> extends MessagingGatewaySupport implements OrderlyShutdownCapable {
+public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implements Pausable, OrderlyShutdownCapable {
 
 	private static final ThreadLocal<AttributeAccessor> ATTRIBUTES_HOLDER = new ThreadLocal<>();
 
@@ -194,6 +195,21 @@ public class KafkaInboundGateway<K, V, R> 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";

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

@@ -339,6 +339,11 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
 		this.messageListenerContainer.resume();
 	}
 
+	@Override
+	public boolean isPaused() {
+		return this.messageListenerContainer.isContainerPaused();
+	}
+
 	@Override
 	public int beforeShutdown() {
 		this.messageListenerContainer.stop();

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageSource.java

@@ -246,15 +246,6 @@ public class KafkaMessageSource<K, V> extends AbstractMessageSource<Object> 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<K, V> extends AbstractMessageSource<Object> 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) {

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<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
 		ContainerProperties containerProps = new ContainerProperties(topic1);
 		containerProps.setIdleEventInterval(100L);
+		containerProps.setPollTimeout(200L);
 		KafkaMessageListenerContainer<Integer, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
 		ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
 		KafkaTemplate<Integer, String> 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();
 	}

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();
 	}