diff --git a/spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerPausedEvent.java b/spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerPausedEvent.java new file mode 100644 index 00000000..07cec097 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerPausedEvent.java @@ -0,0 +1,54 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.event; + +import java.util.Collection; + +import org.apache.kafka.common.TopicPartition; + +/** + * An event published when a consumer is paused. + * + * @author Gary Russell + * @since 2.1.5 + * + */ +@SuppressWarnings("serial") +public class ConsumerPausedEvent extends KafkaEvent { + + private final Collection partitions; + + /** + * Construct an instance with the provided source and partitions. + * @param source the container. + * @param partitions the partitions. + */ + public ConsumerPausedEvent(Object source, Collection partitions) { + super(source); + this.partitions = partitions; + } + + public Collection getPartitions() { + return this.partitions; + } + + @Override + public String toString() { + return "ConsumerPausedEvent [partitions=" + this.partitions + "]"; + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerResumedEvent.java b/spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerResumedEvent.java new file mode 100644 index 00000000..10c43544 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/event/ConsumerResumedEvent.java @@ -0,0 +1,54 @@ +/* + * Copyright 2018 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. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.event; + +import java.util.Collection; + +import org.apache.kafka.common.TopicPartition; + +/** + * An event published when a consumer is resumed. + * + * @author Gary Russell + * @since 2.1.5 + * + */ +@SuppressWarnings("serial") +public class ConsumerResumedEvent extends KafkaEvent { + + private final Collection partitions; + + /** + * Construct an instance with the provided source and partitions. + * @param source the container. + * @param partitions the partitions. + */ + public ConsumerResumedEvent(Object source, Collection partitions) { + super(source); + this.partitions = partitions; + } + + public Collection getPartitions() { + return this.partitions; + } + + @Override + public String toString() { + return "ConsumerResumedEvent [partitions=" + this.partitions + "]"; + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/event/ListenerContainerIdleEvent.java b/spring-kafka/src/main/java/org/springframework/kafka/event/ListenerContainerIdleEvent.java index d5eee42c..d9284f06 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/event/ListenerContainerIdleEvent.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/event/ListenerContainerIdleEvent.java @@ -1,5 +1,5 @@ /* - * Copyright 2016-2017 the original author or authors. + * Copyright 2016-2018 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. @@ -40,15 +40,44 @@ public class ListenerContainerIdleEvent extends KafkaEvent { private final List topicPartitions; + private final boolean paused; + private transient Consumer consumer; + /** + * Construct an instance with the provided arguments. + * @param source the container. + * @param idleTime the idle time. + * @param id the container id. + * @param topicPartitions the topics/partitions currently assigned. + * @param consumer the consumer. + * @deprecated in favor of + * {@link #ListenerContainerIdleEvent(Object, long, String, Collection, Consumer, boolean)} + */ + @Deprecated public ListenerContainerIdleEvent(Object source, long idleTime, String id, Collection topicPartitions, Consumer consumer) { + this(source, idleTime, id, topicPartitions, consumer, false); + } + + /** + * Construct an instance with the provided arguments. + * @param source the container. + * @param idleTime the idle time. + * @param id the container id. + * @param topicPartitions the topics/partitions currently assigned. + * @param consumer the consumer. + * @param paused true if the consumer is paused. + * @since 2.1.5 + */ + public ListenerContainerIdleEvent(Object source, long idleTime, String id, + Collection topicPartitions, Consumer consumer, boolean paused) { super(source); this.idleTime = idleTime; this.listenerId = id; this.topicPartitions = topicPartitions == null ? null : new ArrayList<>(topicPartitions); this.consumer = consumer; + this.paused = paused; } /** @@ -85,11 +114,21 @@ public class ListenerContainerIdleEvent extends KafkaEvent { return this.consumer; } + /** + * Return true if the consumer was paused at the time the idle event was published. + * @return paused. + * @since 2.1.5 + */ + public boolean isPaused() { + return this.paused; + } + @Override public String toString() { return "ListenerContainerIdleEvent [idleTime=" + ((float) this.idleTime / 1000) + "s, listenerId=" + this.listenerId + ", container=" + getSource() + + ", paused=" + this.paused + ", topicPartitions=" + this.topicPartitions + "]"; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java index ff3c3bb6..146c2e88 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/AbstractMessageListenerContainer.java @@ -193,6 +193,11 @@ public abstract class AbstractMessageListenerContainer return this.paused; } + @Override + public boolean isPauseRequested() { + return this.paused; + } + public void setPhase(int phase) { this.phase = phase; } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java index bf5c10bf..c9d97a0f 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java @@ -109,6 +109,19 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis return assigned; } + @Override + public boolean isContainerPaused() { + boolean paused = isPaused(); + if (paused) { + for (AbstractMessageListenerContainer container : this.containers) { + if (!container.isContainerPaused()) { + return false; + } + } + } + return paused; + } + @Override public Map> metrics() { Map> metrics = new HashMap<>(); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java index c722af92..448d88a3 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java @@ -55,6 +55,8 @@ import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.core.KafkaResourceHolder; import org.springframework.kafka.core.ProducerFactoryUtils; +import org.springframework.kafka.event.ConsumerPausedEvent; +import org.springframework.kafka.event.ConsumerResumedEvent; import org.springframework.kafka.event.ListenerContainerIdleEvent; import org.springframework.kafka.event.NonResponsiveConsumerEvent; import org.springframework.kafka.listener.ConsumerSeekAware.ConsumerSeekCallback; @@ -204,6 +206,11 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } + @Override + public boolean isContainerPaused() { + return isPaused() && this.listenerConsumer.consumerPaused; + } + @Override public Map> metrics() { ListenerConsumer listenerConsumer = this.listenerConsumer; @@ -288,21 +295,35 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } - private void publishIdleContainerEvent(long idleTime, Consumer consumer) { + private void publishIdleContainerEvent(long idleTime, Consumer consumer, boolean paused) { if (getApplicationEventPublisher() != null) { getApplicationEventPublisher().publishEvent(new ListenerContainerIdleEvent( - KafkaMessageListenerContainer.this, idleTime, getBeanName(), getAssignedPartitions(), consumer)); + this, idleTime, getBeanName(), getAssignedPartitions(), consumer, paused)); } } private void publishNonResponsiveConsumerEvent(long timeSinceLastPoll, Consumer consumer) { if (getApplicationEventPublisher() != null) { getApplicationEventPublisher().publishEvent( - new NonResponsiveConsumerEvent(KafkaMessageListenerContainer.this, timeSinceLastPoll, + new NonResponsiveConsumerEvent(this, timeSinceLastPoll, getBeanName(), getAssignedPartitions(), consumer)); } } + private void publishConsumerPausedEvent(Collection partitions) { + if (getApplicationEventPublisher() != null) { + getApplicationEventPublisher().publishEvent(new ConsumerPausedEvent(this, + Collections.unmodifiableCollection(partitions))); + } + } + + private void publishConsumerResumedEvent(Collection partitions) { + if (getApplicationEventPublisher() != null) { + getApplicationEventPublisher().publishEvent(new ConsumerResumedEvent(this, + Collections.unmodifiableCollection(partitions))); + } + } + @Override public String toString() { return "KafkaMessageListenerContainer [id=" + getBeanName() @@ -671,14 +692,17 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener if (this.logger.isDebugEnabled()) { this.logger.debug("Paused consumption from: " + this.consumer.paused()); } + publishConsumerPausedEvent(this.consumer.assignment()); } ConsumerRecords records = this.consumer.poll(this.containerProperties.getPollTimeout()); if (this.consumerPaused && !isPaused()) { if (this.logger.isDebugEnabled()) { this.logger.debug("Resuming consumption from: " + this.consumer.paused()); } - this.consumer.resume(this.consumer.paused()); + Set paused = this.consumer.paused(); + this.consumer.resume(paused); this.consumerPaused = false; + publishConsumerResumedEvent(paused); } if (records != null && this.logger.isDebugEnabled()) { this.logger.debug("Received: " + records.count() + " records"); @@ -702,7 +726,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener if (now > lastReceive + this.containerProperties.getIdleEventInterval() && now > lastAlertAt + this.containerProperties.getIdleEventInterval()) { publishIdleContainerEvent(now - lastReceive, this.isConsumerAwareListener - ? this.consumer : null); + ? this.consumer : null, this.consumerPaused); lastAlertAt = now; if (this.genericListener instanceof ConsumerSeekAware) { seekPartitions(getAssignedPartitions(), true); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java index c473b4c9..c9cc096f 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java @@ -85,4 +85,24 @@ public interface MessageListenerContainer extends SmartLifecycle { throw new UnsupportedOperationException("This container doesn't support resume"); } + /** + * Return true if {@link #pause()} has been called; the container might not have actually + * paused yet. + * @return true if pause has been requested. + * @since 2.1.5 + */ + default boolean isPauseRequested() { + throw new UnsupportedOperationException("This container doesn't support pause/resume"); + } + + /** + * Return true if {@link #pause()} has been called; and all consumers in this container + * have actually paused. + * @return true if the container is paused. + * @since 2.1.5 + */ + default boolean isContainerPaused() { + throw new UnsupportedOperationException("This container doesn't support pause/resume"); + } + } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index 7595860a..6bbe83ad 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java @@ -73,6 +73,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.event.NonResponsiveConsumerEvent; import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode; import org.springframework.kafka.listener.adapter.FilteringMessageListenerAdapter; @@ -1649,13 +1651,13 @@ public class KafkaMessageListenerContainerTests { return null; }).given(consumer).commitSync(any(Map.class)); given(consumer.assignment()).willReturn(records.keySet()); - final CountDownLatch pauseLatch = new CountDownLatch(1); + final CountDownLatch pauseLatch = new CountDownLatch(2); willAnswer(i -> { pauseLatch.countDown(); return null; }).given(consumer).pause(records.keySet()); given(consumer.paused()).willReturn(records.keySet()); - final CountDownLatch resumeLatch = new CountDownLatch(1); + final CountDownLatch resumeLatch = new CountDownLatch(2); willAnswer(i -> { resumeLatch.countDown(); return null; @@ -1669,6 +1671,14 @@ public class KafkaMessageListenerContainerTests { containerProps.setMessageListener((MessageListener) r -> { }); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); + container.setApplicationEventPublisher(e -> { + if (e instanceof ConsumerPausedEvent) { + pauseLatch.countDown(); + } + else if (e instanceof ConsumerResumedEvent) { + resumeLatch.countDown(); + } + }); container.start(); assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue(); verify(consumer, times(2)).commitSync(any(Map.class)); diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 84dc73bc..8ea136ef 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -1273,7 +1273,7 @@ In addition, if the broker is unreachable (at the time of writing), the consumer To solve this issue, the container will publish a `NonResponsiveConsumerEvent` if a poll does not return within 3x the `pollInterval` property. By default, this check is performed once every 30 seconds in each container. You can modify the behavior by setting the `monitorInterval` and `noPollThreshold` properties in the `ContainerProperties` when configuring the listener container. -Receiveing such an event will allow you to stop the container(s), thus waking the consumer so it can terminate. +Receiving such an event will allow you to stop the container(s), thus waking the consumer so it can terminate. ====== Event Consumption @@ -1293,7 +1293,9 @@ The events have 5 properties: - `topicPartitions` - the topics/partitions that the container was assigned at the time the event was generated - `consumer` - a reference to the kafka `Consumer` object; for example, if the consumer was previously `pause()` d, it can be `resume()` d when the event is received. -The event is published on the consumer thread, so it is safe to interact with the `Consumer` object. +Starting with _version 2.1.5_, the idle event has a boolean property `paused` which indicates whether the consumer is currently paused; see <> for more information. + +The event is normally published on the consumer thread, so it is safe to interact with the `Consumer` object. [source, xml] ---- @@ -1383,6 +1385,11 @@ To safely pause/resume consumers, you should use the methods on the listener con `pause()` takes effect just before the next `poll()`; `resume` takes effect, just after the current `poll()` returns. When a container is paused, it continues to `poll()` the consumer, avoiding a rebalance if group management is being used, but will not retrieve any records; refer to the Kafka documentation for more information. +Starting with _version 2.1.5_, you can call `isPauseRequested()` to see if `pause()` has been called. +However, the consumers might not have actually paused yet; `isConsumerPaused()` will return true if all `Consumer` s have actually paused. + +In addition, also since _2.1.5_, `ConsumerPausedEvent` s and `ConsumerResumedEvent` s are published with the container as the `source` property and the `TopicPatition` s involved in the `partitions` s property. + [[serdes]] ==== Serialization/Deserialization and Message Conversion