diff --git a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java index b6e982c7..a6d554c1 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java @@ -18,6 +18,8 @@ package org.springframework.kafka.config; import org.springframework.beans.BeanUtils; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.kafka.core.ConsumerFactory; import org.springframework.kafka.listener.AbstractMessageListenerContainer; import org.springframework.kafka.listener.adapter.DeDuplicationStrategy; @@ -32,11 +34,12 @@ import org.springframework.kafka.support.converter.MessageConverter; * @param the value type. * * @author Stephane Nicoll + * @author Gary Russell * * @see AbstractMessageListenerContainer */ public abstract class AbstractKafkaListenerContainerFactory, K, V> - implements KafkaListenerContainerFactory { + implements KafkaListenerContainerFactory, ApplicationEventPublisherAware { private final ContainerProperties containerProperties = new ContainerProperties("propertiesFactory"); @@ -50,6 +53,8 @@ public abstract class AbstractKafkaListenerContainerFactory deDuplicationStrategy; + private ApplicationEventPublisher applicationEventPublisher; + /** * Specify a {@link ConsumerFactory} to use. * @param consumerFactory The consumer factory. @@ -96,6 +101,11 @@ public abstract class AbstractKafkaListenerContainerFactory listenerContainers = getListenerContainers(); AggregatingCallback aggregatingCallback = new AggregatingCallback(listenerContainers.size(), callback); for (MessageListenerContainer listenerContainer : listenerContainers) { - listenerContainer.stop(aggregatingCallback); + if (listenerContainer.isRunning()) { + listenerContainer.stop(aggregatingCallback); + } + else { + aggregatingCallback.run(); + } } } @@ -297,7 +302,7 @@ public class KafkaListenerEndpointRegistry implements DisposableBean, SmartLifec @Override public void run() { - if (this.count.decrementAndGet() == 0) { + if (this.count.decrementAndGet() <= 0) { this.finishCallback.run(); } } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/event/KafkaEvent.java b/spring-kafka/src/main/java/org/springframework/kafka/event/KafkaEvent.java new file mode 100644 index 00000000..b0dad002 --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/event/KafkaEvent.java @@ -0,0 +1,35 @@ +/* + * Copyright 2015-2016 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 org.springframework.context.ApplicationEvent; + + +/** + * Base class for events. + * + * @author Gary Russell + * + */ +@SuppressWarnings("serial") +public abstract class KafkaEvent extends ApplicationEvent { + + public KafkaEvent(Object source) { + super(source); + } + +} 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 new file mode 100644 index 00000000..30f8e21e --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/event/ListenerContainerIdleEvent.java @@ -0,0 +1,82 @@ +/* + * Copyright 2016 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.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.List; + +import org.apache.kafka.common.TopicPartition; + +/** + * An event that is emitted when a container is idle if the container + * is configured to do so. + * + * @author Gary Russell + * + */ +@SuppressWarnings("serial") +public class ListenerContainerIdleEvent extends KafkaEvent { + + private final long idleTime; + + private final String listenerId; + + private final List topicPartitions; + + public ListenerContainerIdleEvent(Object source, long idleTime, String id, + Collection topicPartitions) { + super(source); + this.idleTime = idleTime; + this.listenerId = id; + this.topicPartitions = new ArrayList<>(topicPartitions); + } + + /** + * How long the container has been idle. + * @return the time in milliseconds. + */ + public long getIdleTime() { + return this.idleTime; + } + + /** + * The TopicPartitions the container is listening to. + * @return the TopicPartition list. + */ + public Collection getTopicPartitions() { + return Collections.unmodifiableList(this.topicPartitions); + } + + /** + * The id of the listener (if {@code @RabbitListener}) or the container bean name. + * @return the id. + */ + public String getListenerId() { + return this.listenerId; + } + + @Override + public String toString() { + return "ListenerContainerIdleEvent [idleTime=" + + ((float) this.idleTime / 1000) + "s, listenerId=" + this.listenerId + + ", container=" + getSource() + + ", topicPartitions=" + this.topicPartitions + "]"; + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/event/package-info.java b/spring-kafka/src/main/java/org/springframework/kafka/event/package-info.java new file mode 100644 index 00000000..79d138bc --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/event/package-info.java @@ -0,0 +1,4 @@ +/** + * Application Events. + */ +package org.springframework.kafka.event; 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 373c5d3c..deb7441c 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 @@ -28,6 +28,8 @@ import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.common.TopicPartition; import org.springframework.beans.factory.BeanNameAware; +import org.springframework.context.ApplicationEventPublisher; +import org.springframework.context.ApplicationEventPublisherAware; import org.springframework.context.SmartLifecycle; import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.retry.RecoveryCallback; @@ -44,7 +46,7 @@ import org.springframework.util.Assert; * @author Marius Bogoevici */ public abstract class AbstractMessageListenerContainer - implements MessageListenerContainer, BeanNameAware, SmartLifecycle { + implements MessageListenerContainer, BeanNameAware, ApplicationEventPublisherAware, SmartLifecycle { protected final Log logger = LogFactory.getLog(this.getClass()); // NOSONAR @@ -107,6 +109,8 @@ public abstract class AbstractMessageListenerContainer private String beanName; + private ApplicationEventPublisher applicationEventPublisher; + private boolean autoStartup = true; private int phase = 0; @@ -116,8 +120,8 @@ public abstract class AbstractMessageListenerContainer protected AbstractMessageListenerContainer(ContainerProperties containerProperties) { Assert.notNull(containerProperties, "'containerProperties' cannot be null"); this.containerProperties = containerProperties; - if (containerProperties.consumerRebalanceListener == null) { - containerProperties.consumerRebalanceListener = createConsumerRebalanceListener(); + if (containerProperties.getConsumerRebalanceListener() == null) { + containerProperties.setConsumerRebalanceListener(createConsumerRebalanceListener()); } } @@ -130,6 +134,15 @@ public abstract class AbstractMessageListenerContainer return this.beanName; } + @Override + public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) { + this.applicationEventPublisher = applicationEventPublisher; + } + + public ApplicationEventPublisher getApplicationEventPublisher() { + return this.applicationEventPublisher; + } + @Override public boolean isAutoStartup() { return this.autoStartup; @@ -163,28 +176,28 @@ public abstract class AbstractMessageListenerContainer @Override public void setupMessageListener(Object messageListener) { - this.containerProperties.messageListener = messageListener; + this.containerProperties.setMessageListener(messageListener); } @Override public final void start() { synchronized (this.lifecycleMonitor) { Assert.isTrue( - this.containerProperties.messageListener instanceof MessageListener - || this.containerProperties.messageListener instanceof AcknowledgingMessageListener, + this.containerProperties.getMessageListener() instanceof MessageListener + || this.containerProperties.getMessageListener() instanceof AcknowledgingMessageListener, "Either a " + MessageListener.class.getName() + " or a " + AcknowledgingMessageListener.class.getName() + " must be provided"); - if (this.containerProperties.recoveryCallback == null) { - this.containerProperties.recoveryCallback = new RecoveryCallback() { + if (this.containerProperties.getRecoveryCallback() == null) { + this.containerProperties.setRecoveryCallback(new RecoveryCallback() { @Override public Void recover(RetryContext context) throws Exception { @SuppressWarnings("unchecked") ConsumerRecord record = (ConsumerRecord) context.getAttribute("record"); Throwable lastThrowable = context.getLastThrowable(); - if (AbstractMessageListenerContainer.this.containerProperties.errorHandler != null + if (AbstractMessageListenerContainer.this.containerProperties.getErrorHandler() != null && lastThrowable instanceof Exception) { - AbstractMessageListenerContainer.this.containerProperties.errorHandler + AbstractMessageListenerContainer.this.containerProperties.getErrorHandler() .handle((Exception) lastThrowable, record); } else { @@ -194,7 +207,7 @@ public abstract class AbstractMessageListenerContainer return null; } - }; + }); } doStart(); } @@ -212,7 +225,7 @@ public abstract class AbstractMessageListenerContainer } }); try { - latch.await(this.containerProperties.shutdownTimeout, TimeUnit.MILLISECONDS); + latch.await(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS); } catch (InterruptedException e) { } 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 1e9cef2e..258c086f 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 @@ -20,6 +20,7 @@ import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; import org.apache.kafka.common.TopicPartition; @@ -96,18 +97,19 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis protected void doStart() { if (!isRunning()) { ContainerProperties containerProperties = getContainerProperties(); - if (containerProperties.topicPartitions != null - && this.concurrency > containerProperties.topicPartitions.length) { + TopicPartition[] topicPartitions = containerProperties.getTopicPartitions(); + if (topicPartitions != null + && this.concurrency > topicPartitions.length) { this.logger.warn("When specific partitions are provided, the concurrency must be less than or " + "equal to the number of partitions; reduced from " + this.concurrency + " to " - + containerProperties.topicPartitions.length); - this.concurrency = containerProperties.topicPartitions.length; + + topicPartitions.length); + this.concurrency = topicPartitions.length; } setRunning(true); for (int i = 0; i < this.concurrency; i++) { KafkaMessageListenerContainer container; - if (containerProperties.topicPartitions == null) { + if (topicPartitions == null) { container = new KafkaMessageListenerContainer<>(this.consumerFactory, containerProperties); } else { @@ -117,6 +119,9 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis if (getBeanName() != null) { container.setBeanName(getBeanName() + "-" + i); } + if (getApplicationEventPublisher() != null) { + container.setApplicationEventPublisher(getApplicationEventPublisher()); + } container.start(); this.containers.add(container); } @@ -124,24 +129,23 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis } private TopicPartition[] partitionSubset(ContainerProperties containerProperties, int i) { + TopicPartition[] topicPartitions = containerProperties.getTopicPartitions(); if (this.concurrency == 1) { - return containerProperties.topicPartitions; + return topicPartitions; } else { - int numPartitions = containerProperties.topicPartitions.length; + int numPartitions = topicPartitions.length; if (numPartitions == this.concurrency) { - return new TopicPartition[] { containerProperties.topicPartitions[i] }; + return new TopicPartition[] { topicPartitions[i] }; } else { int perContainer = numPartitions / this.concurrency; TopicPartition[] subset; if (i == this.concurrency - 1) { - subset = Arrays.copyOfRange(containerProperties.topicPartitions, i * perContainer, - containerProperties.topicPartitions.length); + subset = Arrays.copyOfRange(topicPartitions, i * perContainer, topicPartitions.length); } else { - subset = Arrays.copyOfRange(containerProperties.topicPartitions, i * perContainer, - (i + 1) * perContainer); + subset = Arrays.copyOfRange(topicPartitions, i * perContainer, (i + 1) * perContainer); } return subset; } @@ -152,14 +156,37 @@ public class ConcurrentMessageListenerContainer extends AbstractMessageLis * Under lifecycle lock. */ @Override - protected void doStop(Runnable callback) { + protected void doStop(final Runnable callback) { + final AtomicInteger count = new AtomicInteger(); if (isRunning()) { setRunning(false); for (KafkaMessageListenerContainer container : this.containers) { - container.stop(callback); + if (container.isRunning()) { + count.incrementAndGet(); + } + } + for (KafkaMessageListenerContainer container : this.containers) { + if (container.isRunning()) { + container.stop(new Runnable() { + + @Override + public void run() { + if (count.decrementAndGet() <= 0) { + callback.run(); + } + } + + }); + } } this.containers.clear(); } } + @Override + public String toString() { + return "ConcurrentMessageListenerContainer [concurrency=" + this.concurrency + ", beanName=" + + this.getBeanName() + ", running=" + this.isRunning() + "]"; + } + } 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 0e024f8c..fa380b9e 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 @@ -44,9 +44,11 @@ import org.apache.kafka.clients.consumer.OffsetCommitCallback; import org.apache.kafka.common.TopicPartition; import org.apache.kafka.common.errors.WakeupException; +import org.springframework.context.ApplicationEventPublisher; import org.springframework.core.task.SimpleAsyncTaskExecutor; import org.springframework.kafka.KafkaException; import org.springframework.kafka.core.ConsumerFactory; +import org.springframework.kafka.event.ListenerContainerIdleEvent; import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.kafka.support.Acknowledgment; import org.springframework.retry.RetryCallback; @@ -111,7 +113,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener this.topicPartitions = Arrays.asList(topicPartitions).toArray(new TopicPartition[topicPartitions.length]); } else { - this.topicPartitions = containerProperties.topicPartitions; + this.topicPartitions = containerProperties.getTopicPartitions(); } } @@ -140,7 +142,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener return; } setRunning(true); - Object messageListener = getContainerProperties().messageListener; + Object messageListener = getContainerProperties().getMessageListener(); Assert.state(messageListener != null, "A MessageListener is required"); if (messageListener instanceof AcknowledgingMessageListener) { this.acknowledgingMessageListener = (AcknowledgingMessageListener) messageListener; @@ -152,19 +154,19 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener throw new IllegalStateException("messageListener must be 'MessageListener' " + "or 'AcknowledgingMessageListener', not " + messageListener.getClass().getName()); } - if (getContainerProperties().consumerTaskExecutor == null) { + if (getContainerProperties().getConsumerTaskExecutor() == null) { SimpleAsyncTaskExecutor consumerExecutor = new SimpleAsyncTaskExecutor( (getBeanName() == null ? "" : getBeanName()) + "-kafka-consumer-"); - getContainerProperties().consumerTaskExecutor = consumerExecutor; + getContainerProperties().setConsumerTaskExecutor(consumerExecutor); } - if (getContainerProperties().listenerTaskExecutor == null) { + if (getContainerProperties().getListenerTaskExecutor() == null) { SimpleAsyncTaskExecutor listenerExecutor = new SimpleAsyncTaskExecutor( (getBeanName() == null ? "" : getBeanName()) + "-kafka-listener-"); - getContainerProperties().listenerTaskExecutor = listenerExecutor; + getContainerProperties().setListenerTaskExecutor(listenerExecutor); } this.listenerConsumer = new ListenerConsumer(this.listener, this.acknowledgingMessageListener, - getContainerProperties().recentOffset); - this.listenerConsumerFuture = getContainerProperties().consumerTaskExecutor + getContainerProperties().getRecentOffset()); + this.listenerConsumerFuture = getContainerProperties().getConsumerTaskExecutor() .submitListenable(this.listenerConsumer); } @@ -182,6 +184,10 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener @Override public void onSuccess(Object result) { + if (KafkaMessageListenerContainer.this.logger.isDebugEnabled()) { + KafkaMessageListenerContainer.this.logger + .debug(KafkaMessageListenerContainer.this + " stopped normally"); + } if (callback != null) { callback.run(); } @@ -192,14 +198,22 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } + + @Override + public String toString() { + return "KafkaMessageListenerContainer [id=" + getBeanName() + ", topicPartitions=" + getAssignedPartitions() + + "]"; + } + + private final class ListenerConsumer implements SchedulingAwareRunnable { private final Log logger = LogFactory.getLog(ListenerConsumer.class); private final ContainerProperties containerProperties = getContainerProperties(); - private final OffsetCommitCallback commitCallback = this.containerProperties.commitCallback != null - ? this.containerProperties.commitCallback + private final OffsetCommitCallback commitCallback = this.containerProperties.getCommitCallback() != null + ? this.containerProperties.getCommitCallback() : new LoggingCommitCallback(); private final Consumer consumer; @@ -216,20 +230,23 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private final boolean autoCommit = KafkaMessageListenerContainer.this.consumerFactory.isAutoCommit(); - private final boolean isManualAck = this.containerProperties.ackMode.equals(AckMode.MANUAL); + private final boolean isManualAck = this.containerProperties.getAckMode().equals(AckMode.MANUAL); - private final boolean isManualImmediateAck = this.containerProperties.ackMode.equals(AckMode.MANUAL_IMMEDIATE) - || this.containerProperties.ackMode.equals(AckMode.MANUAL_IMMEDIATE_SYNC); + private final boolean isManualImmediateAck = + this.containerProperties.getAckMode().equals(AckMode.MANUAL_IMMEDIATE) + || this.containerProperties.getAckMode().equals(AckMode.MANUAL_IMMEDIATE_SYNC); private final boolean isAnyManualAck = this.isManualAck || this.isManualImmediateAck; - private final boolean isRecordAck = this.containerProperties.ackMode.equals(AckMode.RECORD); + private final boolean isRecordAck = this.containerProperties.getAckMode().equals(AckMode.RECORD); - private final BlockingQueue> recordsToProcess = new LinkedBlockingQueue<>( - this.containerProperties.queueDepth); + private final BlockingQueue> recordsToProcess = + new LinkedBlockingQueue<>(this.containerProperties.getQueueDepth()); private final BlockingQueue> acks = new LinkedBlockingQueue<>(); + private final ApplicationEventPublisher applicationEventPublisher = getApplicationEventPublisher(); + private volatile Collection definedPartitions; private volatile Collection assignedPartitions; @@ -252,7 +269,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private ListenerConsumer(MessageListener listener, AcknowledgingMessageListener ackListener, long recentOffset) { Assert.state(!this.isAnyManualAck || !this.autoCommit, - "Consumer cannot be configured for auto commit for ackMode " + this.containerProperties.ackMode); + "Consumer cannot be configured for auto commit for ackMode " + this.containerProperties.getAckMode()); Consumer consumer = KafkaMessageListenerContainer.this.consumerFactory.createConsumer(); ConsumerRebalanceListener rebalanceListener = new ConsumerRebalanceListener() { @@ -284,7 +301,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener + "autocommit mode, so transition will be handled by the consumer"); } } - getContainerProperties().consumerRebalanceListener.onPartitionsRevoked(partitions); + getContainerProperties().getConsumerRebalanceListener().onPartitionsRevoked(partitions); } @Override @@ -299,17 +316,17 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener && !CollectionUtils.isEmpty(partitions)) { startInvoker(); } - getContainerProperties().consumerRebalanceListener.onPartitionsAssigned(partitions); + getContainerProperties().getConsumerRebalanceListener().onPartitionsAssigned(partitions); } }; if (KafkaMessageListenerContainer.this.topicPartitions == null) { - if (this.containerProperties.topicPattern != null) { - consumer.subscribe(this.containerProperties.topicPattern, rebalanceListener); + if (this.containerProperties.getTopicPattern() != null) { + consumer.subscribe(this.containerProperties.getTopicPattern(), rebalanceListener); } else { - consumer.subscribe(Arrays.asList(this.containerProperties.topics), rebalanceListener); + consumer.subscribe(Arrays.asList(this.containerProperties.getTopics()), rebalanceListener); } } else { @@ -326,7 +343,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private void startInvoker() { ListenerConsumer.this.invoker = new ListenerInvoker(); - ListenerConsumer.this.listenerInvokerFuture = this.containerProperties.listenerTaskExecutor + ListenerConsumer.this.listenerInvokerFuture = this.containerProperties.getListenerTaskExecutor() .submit(ListenerConsumer.this.invoker); } @@ -349,16 +366,21 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } ConsumerRecords unsent = null; + long lastReceive = System.currentTimeMillis(); + long lastAlertAt = lastReceive; while (isRunning()) { try { if (this.logger.isTraceEnabled()) { this.logger.trace("Polling (paused=" + this.paused + ")..."); } - ConsumerRecords records = this.consumer.poll(this.containerProperties.pollTimeout); + ConsumerRecords records = this.consumer.poll(this.containerProperties.getPollTimeout()); if (this.logger.isDebugEnabled()) { this.logger.debug("Received: " + records.count() + " records"); } if (records != null && records.count() > 0) { + if (this.containerProperties.getIdleEventInterval() != null) { + lastReceive = System.currentTimeMillis(); + } handleManualAcks(); // if the container is set to auto-commit, then execute in the // same thread @@ -379,6 +401,16 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } } + else { + if (this.containerProperties.getIdleEventInterval() != null) { + long now = System.currentTimeMillis(); + if (now > lastReceive + this.containerProperties.getIdleEventInterval() + && now > lastAlertAt + this.containerProperties.getIdleEventInterval()) { + publishIdleContainerEvent(now - lastReceive); + lastAlertAt = now; + } + } + } unsent = checkPause(unsent); if (!this.paused && !this.autoCommit) { processCommits(); @@ -388,8 +420,8 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener unsent = checkPause(unsent); } catch (Exception e) { - if (this.containerProperties.errorHandler != null) { - this.containerProperties.errorHandler.handle(e, null); + if (this.containerProperties.getErrorHandler() != null) { + this.containerProperties.getErrorHandler().handle(e, null); } else { this.logger.error("Container exception", e); @@ -411,10 +443,17 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } + private void publishIdleContainerEvent(long idleTime) { + if (this.applicationEventPublisher != null) { + this.applicationEventPublisher.publishEvent(new ListenerContainerIdleEvent( + KafkaMessageListenerContainer.this, idleTime, getBeanName(), getAssignedPartitions())); + } + } + private void stopInvokerAndCommitManualAcks() { this.invoker.stop(); try { - this.listenerInvokerFuture.get(this.containerProperties.shutdownTimeout, TimeUnit.MILLISECONDS); + this.listenerInvokerFuture.get(this.containerProperties.getShutdownTimeout(), TimeUnit.MILLISECONDS); } catch (InterruptedException e) { Thread.currentThread().interrupt(); @@ -439,7 +478,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } private ConsumerRecords checkPause(ConsumerRecords unsent) { - if (this.paused && this.recordsToProcess.size() < this.containerProperties.queueDepth) { + if (this.paused && this.recordsToProcess.size() < this.containerProperties.getQueueDepth()) { // Listener has caught up. this.consumer.resume( this.assignedPartitions.toArray(new TopicPartition[this.assignedPartitions.size()])); @@ -459,8 +498,8 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } private boolean sendToListener(final ConsumerRecords records) throws InterruptedException { - if (this.containerProperties.pauseEnabled && CollectionUtils.isEmpty(this.definedPartitions)) { - return !this.recordsToProcess.offer(records, this.containerProperties.pauseAfter, + if (this.containerProperties.isPauseEnabled() && CollectionUtils.isEmpty(this.definedPartitions)) { + return !this.recordsToProcess.offer(records, this.containerProperties.getPauseAfter(), TimeUnit.MILLISECONDS); } else { @@ -503,7 +542,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener if (ListenerConsumer.this.logger.isDebugEnabled()) { ListenerConsumer.this.logger.debug("Committing: " + commits); } - if (ListenerConsumer.this.containerProperties.ackMode.equals(AckMode.MANUAL_IMMEDIATE)) { + if (ListenerConsumer.this.containerProperties.getAckMode().equals(AckMode.MANUAL_IMMEDIATE)) { ListenerConsumer.this.consumer.commitAsync(commits, ListenerConsumer.this.commitCallback); } @@ -521,8 +560,9 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } try { if (this.acknowledgingMessageListener != null) { - if (this.containerProperties.retryTemplate != null) { - this.containerProperties.retryTemplate.execute(new RetryCallback() { + if (this.containerProperties.getRetryTemplate() != null) { + this.containerProperties.getRetryTemplate().execute( + new RetryCallback() { @Override public Void doWithRetry(RetryContext context) throws KafkaException { @@ -532,15 +572,16 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener return null; } - }, this.containerProperties.recoveryCallback); + }, this.containerProperties.getRecoveryCallback()); } else { this.acknowledgingMessageListener.onMessage(record, new ConsumerAcknowledgment(record)); } } else { - if (this.containerProperties.retryTemplate != null) { - this.containerProperties.retryTemplate.execute(new RetryCallback() { + if (this.containerProperties.getRetryTemplate() != null) { + this.containerProperties.getRetryTemplate().execute( + new RetryCallback() { @Override public Void doWithRetry(RetryContext context) throws KafkaException { @@ -549,7 +590,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener return null; } - }, this.containerProperties.recoveryCallback); + }, this.containerProperties.getRecoveryCallback()); } else { this.listener.onMessage(record); @@ -561,8 +602,8 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } } catch (Exception e) { - if (this.containerProperties.errorHandler != null) { - this.containerProperties.errorHandler.handle(e, record); + if (this.containerProperties.getErrorHandler() != null) { + this.containerProperties.getErrorHandler().handle(e, record); } else { this.logger.error("Listener threw an exception and no error handler for " + record, e); @@ -574,19 +615,19 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener private void processCommits() { this.count += this.acks.size(); long now; - AckMode ackMode = this.containerProperties.ackMode; + AckMode ackMode = this.containerProperties.getAckMode(); if (!this.isManualImmediateAck) { if (!this.isManualAck) { updatePendingOffsets(); } - boolean countExceeded = this.count >= this.containerProperties.ackCount; + boolean countExceeded = this.count >= this.containerProperties.getAckCount(); if (ackMode.equals(AckMode.BATCH) || ackMode.equals(AckMode.COUNT) && countExceeded) { commitIfNecessary(); this.count = 0; } else { now = System.currentTimeMillis(); - boolean elapsed = now - this.last > this.containerProperties.ackTime; + boolean elapsed = now - this.last > this.containerProperties.getAckTime(); if (ackMode.equals(AckMode.TIME) && elapsed) { commitIfNecessary(); this.last = now; @@ -669,7 +710,7 @@ public class KafkaMessageListenerContainer extends AbstractMessageListener } if (!commits.isEmpty()) { try { - if (this.containerProperties.syncCommits) { + if (this.containerProperties.isSyncCommits()) { this.consumer.commitSync(commits); } else { diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java index f9727692..3919234f 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/config/ContainerProperties.java @@ -51,17 +51,17 @@ public class ContainerProperties { /** * Topic names. */ - public final String[] topics; + private final String[] topics; /** * Topic pattern. */ - public final Pattern topicPattern; + private final Pattern topicPattern; /** * Topics/partitions. */ - public final TopicPartition[] topicPartitions; + private final TopicPartition[] topicPartitions; /** * The ack mode to use when auto ack (in the configuration properties) is false. @@ -76,102 +76,102 @@ public class ContainerProperties { * {@link AcknowledgingMessageListener}. * */ - public AbstractMessageListenerContainer.AckMode ackMode = AckMode.BATCH; + private AbstractMessageListenerContainer.AckMode ackMode = AckMode.BATCH; /** * The number of outstanding record count after which offsets should be * committed when {@link AckMode#COUNT} or {@link AckMode#COUNT_TIME} is being * used. */ - public int ackCount; + private int ackCount; /** * The time (ms) after which outstanding offsets should be committed when * {@link AckMode#TIME} or {@link AckMode#COUNT_TIME} is being used. Should be * larger than */ - public long ackTime; + private long ackTime; /** * The message listener; must be a {@link MessageListener} or * {@link AcknowledgingMessageListener}. */ - public Object messageListener; + private Object messageListener; /** * The max time to block in the consumer waiting for records. */ - public volatile long pollTimeout = 1000; + private volatile long pollTimeout = 1000; /** * The executor for threads that poll the consumer. */ - public AsyncListenableTaskExecutor consumerTaskExecutor; + private AsyncListenableTaskExecutor consumerTaskExecutor; /** * The executor for threads that invoke the listener. */ - public AsyncListenableTaskExecutor listenerTaskExecutor; + private AsyncListenableTaskExecutor listenerTaskExecutor; /** * The error handler to call when the listener throws an exception. */ - public ErrorHandler errorHandler = new LoggingErrorHandler(); + private ErrorHandler errorHandler = new LoggingErrorHandler(); /** * When using Kafka group management and {@link #setPauseEnabled(boolean)} is * true, the delay after which the consumer should be paused. Default 10000. */ - public long pauseAfter = DEFAULT_PAUSE_AFTER; + private long pauseAfter = DEFAULT_PAUSE_AFTER; /** * When true, avoids rebalancing when this consumer is slow or throws a * qualifying exception - pauses the consumer. Default: true. * @see #pauseAfter */ - public boolean pauseEnabled = true; + private boolean pauseEnabled = true; /** * A retry template to retry deliveries. */ - public RetryTemplate retryTemplate; + private RetryTemplate retryTemplate; /** * A recovery callback to be invoked when retries are exhausted. By default * the error handler is invoked. */ - public RecoveryCallback recoveryCallback; + private RecoveryCallback recoveryCallback; /** * Set the queue depth for handoffs from the consumer thread to the listener * thread. Default 1 (up to 2 in process). */ - public int queueDepth = DEFAULT_QUEUE_DEPTH; + private int queueDepth = DEFAULT_QUEUE_DEPTH; /** * The timeout for shutting down the container. This is the maximum amount of * time that the invocation to {@code #stop(Runnable)} will block for, before * returning. */ - public long shutdownTimeout = DEFAULT_SHUTDOWN_TIMEOUT; + private long shutdownTimeout = DEFAULT_SHUTDOWN_TIMEOUT; /** * The offset to this number of records back from the latest when starting. * Overrides any consumer properties (earliest, latest). Only applies when * explicit topic/partition assignment is provided. */ - public long recentOffset; + private long recentOffset; /** * A user defined {@link ConsumerRebalanceListener} implementation. */ - public ConsumerRebalanceListener consumerRebalanceListener; + private ConsumerRebalanceListener consumerRebalanceListener; /** * The commit callback; by default a simple logging callback is used to log * success at DEBUG level and failures at ERROR level. */ - public OffsetCommitCallback commitCallback; + private OffsetCommitCallback commitCallback; /** * Whether or not to call consumer.commitSync() or commitAsync() when the @@ -179,7 +179,9 @@ public class ContainerProperties { * https://github.com/spring-projects/spring-kafka/issues/62 At the time of * writing, async commits are not entirely reliable. */ - public boolean syncCommits = true; + private boolean syncCommits = true; + + private Long idleEventInterval; public ContainerProperties(String... topics) { Assert.notEmpty(topics, "An array of topicPartitions must be provided"); @@ -374,10 +376,14 @@ public class ContainerProperties { this.syncCommits = syncCommits; } - /* - * Although we generally use field access, we need the getters so we can copy - * the properties from the factory when creating an annotated listener. + /** + * Set the idle event interval; when set, an event is emitted if a poll returns + * no records and this interval has elapsed since a record was returned. + * @param idleEventInterval the interval. */ + public void setIdleEventInterval(Long idleEventInterval) { + this.idleEventInterval = idleEventInterval; + } public String[] getTopics() { return this.topics; @@ -463,4 +469,8 @@ public class ContainerProperties { return this.syncCommits; } + public Long getIdleEventInterval() { + return this.idleEventInterval; + } + } diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index bebecd7b..485fecca 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -36,6 +36,7 @@ import org.mockito.Mockito; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.context.event.EventListener; import org.springframework.context.support.PropertySourcesPlaceholderConfigurer; import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory; import org.springframework.kafka.config.KafkaListenerContainerFactory; @@ -45,6 +46,7 @@ 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.ListenerContainerIdleEvent; import org.springframework.kafka.listener.AbstractMessageListenerContainer.AckMode; import org.springframework.kafka.listener.ConcurrentMessageListenerContainer; import org.springframework.kafka.listener.MessageListenerContainer; @@ -124,6 +126,8 @@ public class EnableKafkaIntegrationTests { assertThat(this.listener.latch4.await(20, TimeUnit.SECONDS)).isTrue(); assertThat(this.listener.record.value()).isEqualTo("foo"); assertThat(this.listener.ack).isNotNull(); + assertThat(this.listener.eventLatch.await(20, TimeUnit.SECONDS)).isTrue(); + assertThat(this.listener.event.getListenerId().startsWith("qux-")); template.send("annotated5", 0, 0, "foo"); template.send("annotated5", 1, 0, "bar"); @@ -236,6 +240,7 @@ public class EnableKafkaIntegrationTests { factory.setConsumerFactory(manualConsumerFactory()); ContainerProperties props = factory.getContainerProperties(); props.setAckMode(AckMode.MANUAL_IMMEDIATE); + props.setIdleEventInterval(100L); return factory; } @@ -354,6 +359,8 @@ public class EnableKafkaIntegrationTests { private final CountDownLatch latch7 = new CountDownLatch(1); + private final CountDownLatch eventLatch = new CountDownLatch(1); + private volatile Integer partition; private volatile ConsumerRecord record; @@ -366,6 +373,8 @@ public class EnableKafkaIntegrationTests { private Foo foo; + private volatile ListenerContainerIdleEvent event; + @KafkaListener(id = "manualStart", topics = "manualStart", containerFactory = "kafkaAutoStartFalseListenerContainerFactory") public void manualStart(String foo) { @@ -401,6 +410,12 @@ public class EnableKafkaIntegrationTests { this.latch4.countDown(); } + @EventListener(condition = "event.listenerId.startsWith('qux')") + public void eventHandler(ListenerContainerIdleEvent event) { + this.event = event; + eventLatch.countDown(); + } + @KafkaListener(id = "fiz", topicPartitions = { @TopicPartition(topic = "annotated5", partitions = { "#{'${foo:0,1}'.split(',')}" }), @TopicPartition(topic = "annotated6", partitions = { "0", "1" }) @@ -442,7 +457,7 @@ public class EnableKafkaIntegrationTests { latch1.countDown(); } - @KafkaListener(topics = "annotated9") + @KafkaListener(id = "ifctx", topics = "annotated9") @Transactional public void listenTx(String foo) { latch2.countDown(); @@ -457,7 +472,7 @@ public class EnableKafkaIntegrationTests { } } - @KafkaListener(topics = "annotated8") + @KafkaListener(id = "multi", topics = "annotated8") static class MultiListenerBean { private final CountDownLatch latch1 = new CountDownLatch(1); diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index ba5970a3..263a764e 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -149,7 +149,7 @@ public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, ContainerProperties containerProperties) public KafkaMessageListenerContainer(ConsumerFactory consumerFactory, - ContainerProperties containerProperties, TopicPartition... topicPartitions) { + ContainerProperties containerProperties, TopicPartition... topicPartitions) ---- @@ -228,13 +228,13 @@ NOTE: `MANUAL`, `MANUAL_IMMEDIATE`, and `MANUAL_IMMEDIATE_SYNC` require the list ---- public interface AcknowledgingMessageListener { - void onMessage(ConsumerRecord record, Acknowledgment acknowledgment); + void onMessage(ConsumerRecord record, Acknowledgment acknowledgment); } public interface Acknowledgment { - void acknowledge(); + void acknowledge(); } ---- @@ -409,3 +409,87 @@ public void jsonListener(Foo foo) { ... } ---- + +[[idle-containers]] +===== Detecting Idle Asynchronous Consumers + +While efficient, one problem with asynchronous consumers is detecting when they are idle - users might want to take +some action if no messages arrive for some period of time. + +You can configure the listener container to publish a `ListenerContainerIdleEvent` when some time passes with no message delivery. +While the container is idle, an event will be published every `idleEventInterval` milliseconds. + +To configure this feature, set the `idleEventInterval` on the container: + +[source, java] +---- +@Bean +public KafKaMessageListenerContainer(ConnectionFactory connectionFactory) { + ContainerProperties containerProps = new ContainerProperties("topic1", "topic2"); + ... + containerProps.setIdleEventInterval(60000L); + ... + KafKaMessageListenerContainer container = new KafKaMessageListenerContainer<>(...); + return container; +} +---- + +Or, for a `@KafkaListener`... + +[source, java] +---- +@Bean +public ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() { + ConcurrentKafkaListenerContainerFactory factory = + new ConcurrentKafkaListenerContainerFactory<>(); + ... + factory.getContainerProperties().setIdleEventInterval(60000L); + ... + return factory; +} +---- + +In each of these cases, an event will be published once per minute while the container is idle. + +====== Event Consumption + +You can capture these events by implementing `ApplicationListener` - either a general listener, or one narrowed to only receive this specific event. +You can also use `@EventListener`, introduced in Spring Framework 4.2. + +The following example combines the `@KafkaListener` and `@EventListener` into a single class. +It's important to understand that the application listener will get events for all containers so you may need to +check the listener id if you want to take specific action based on which container is idle. +You can also use the `@EventListener` `condition` for this purpose. + +The events have 4 properties: + +- `source` - the listener container instance +- `id` - the listener id (or container bean name) +- `idleTime` - the time the container had been idle when the event was published +- `topicPartitions` - the topics/partitions that the container was assigned at the time the event was generated + +[source, xml] +---- +public class Listener { + + @KafkaListener(id = "qux", topics = "annotated") + public void listen4(@Payload String foo, Acknowledgment ack) { + ... + } + + @EventListener(condition = "event.listenerId.startsWith('qux-')") + public void eventHandler(ListenerContainerIdleEvent event) { + this.event = event; + eventLatch.countDown(); + } + +} +---- + +IMPORTANT: Event listeners will see events for all containers; so, in the example above, we narrow the events received based on the listener ID. +Since containers created for the `@KafkaListener` support concurrency, the actual containers are named `id-n` where the `n` is a unique value for each instance to support the concurrency. +Hence we use `startsWith` in the condition. + +CAUTION: If you wish to use the idle event to stop the lister container, you should not call `container.stop()` on the thread that calls the listener - it will cause delays and unnecessary log messages. +Instead, you should hand off the event to a different thread that can then stop the container. +Also, you should not `stop()` the container instance in the event if it is a child container, you should stop the concurrent container instead.