From 46d06fd404f65f39784c3a3e8d4ea17c8f87ccc2 Mon Sep 17 00:00:00 2001 From: Artem Bilan Date: Thu, 8 Jan 2015 12:05:26 +0200 Subject: [PATCH] KAFKA-10: Fix for `ConsumerConfiguration.executor` Fixes GH10 https://github.com/spring-projects/spring-integration-kafka/issues/10 Previously the `ConsumerConfiguration` used an internal `ExecutorService`. In the management environment (App Server) it causes an issue with unmanaged Threads. * Add `setExecutor` option and expose it to the Namespace to inject a managed `Executor` * Stop an internal `ExecutorService` manually from the `ConsumerConfiguration.shutdown()` Polishing --- .../xml/KafkaConsumerContextParser.java | 4 +- .../kafka/support/ConsumerConfiguration.java | 77 ++++++++++++++++--- .../kafka/support/KafkaConsumerContext.java | 2 +- .../xml/spring-integration-kafka-1.0.xsd | 26 ++++++- 4 files changed, 92 insertions(+), 17 deletions(-) diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java index fa1700f76e..ffad4fb64e 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/config/xml/KafkaConsumerContextParser.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2014 the original author or authors. + * Copyright 2002-2015 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. @@ -81,6 +81,8 @@ public class KafkaConsumerContextParser extends AbstractSingleBeanDefinitionPars "key-decoder"); IntegrationNamespaceUtils.setValueIfAttributeDefined(consumerConfigurationBuilder, consumerConfiguration, "max-messages"); + IntegrationNamespaceUtils.setReferenceIfAttributeDefined(consumerConfigurationBuilder, consumerConfiguration, + "executor"); IntegrationNamespaceUtils.setValueIfAttributeDefined(consumerMetadataBuilder, parentElem, "consumer-timeout"); diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java index 2514a08650..d5af934b35 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/ConsumerConfiguration.java @@ -1,42 +1,66 @@ /* - * Copyright 2002-2013 the original author or authors. Licensed under the Apache License, Version 2.0 (the "License"); + * Copyright 2002-2015 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.integration.kafka.support; -import java.util.*; -import java.util.concurrent.*; +import java.util.ArrayList; +import java.util.Collection; +import java.util.LinkedList; +import java.util.List; +import java.util.Map; +import java.util.concurrent.Callable; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executor; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.springframework.messaging.MessagingException; +import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor; +import org.springframework.util.Assert; import kafka.consumer.ConsumerTimeoutException; import kafka.consumer.KafkaStream; import kafka.javaapi.consumer.ConsumerConnector; import kafka.message.MessageAndMetadata; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.springframework.messaging.MessagingException; - /** * @author Soby Chacko * @author Rajasekar Elango + * @author Artem Bilan * @since 0.5 */ public class ConsumerConfiguration { private static final Log LOGGER = LogFactory.getLog(ConsumerConfiguration.class); private final ConsumerMetadata consumerMetadata; + private final ConsumerConnectionProvider consumerConnectionProvider; + private final MessageLeftOverTracker messageLeftOverTracker; + private ConsumerConnector consumerConnector; + private volatile int count = 0; + private int maxMessages = 1; + private Collection>> consumerMessageStreams; - private final ExecutorService executorService = Executors.newCachedThreadPool(); + private ExecutorService executorService = Executors.newCachedThreadPool(); + + private boolean executorExplicitlySet; + + private volatile boolean stopped; public ConsumerConfiguration(final ConsumerMetadata consumerMetadata, final ConsumerConnectionProvider consumerConnectionProvider, @@ -46,6 +70,19 @@ public class ConsumerConfiguration { this.messageLeftOverTracker = messageLeftOverTracker; } + public void setExecutor(Executor executor) { + boolean isExecutorService = executor instanceof ExecutorService; + boolean isThreadPoolTaskExecutor = executor instanceof ThreadPoolTaskExecutor; + Assert.isTrue(isExecutorService || isThreadPoolTaskExecutor); + if (isExecutorService) { + this.executorService = (ExecutorService) executor; + } + else { + this.executorService = ((ThreadPoolTaskExecutor) executor).getThreadPoolExecutor(); + } + this.executorExplicitlySet = true; + } + public ConsumerMetadata getConsumerMetadata() { return consumerMetadata; } @@ -75,7 +112,8 @@ public class ConsumerConfiguration { } } } - } catch (ConsumerTimeoutException cte) { + } + catch (ConsumerTimeoutException cte) { LOGGER.debug("Consumer timed out"); } return rawMessages; @@ -93,7 +131,7 @@ public class ConsumerConfiguration { messages.putAll(getLeftOverMessageMap()); try { - for (final Future>> result : executorService.invokeAll(tasks)) { + for (final Future>> result : this.executorService.invokeAll(tasks)) { if (!result.get().isEmpty()) { final String topic = result.get().get(0).topic(); if (!messages.containsKey(topic)) { @@ -106,8 +144,14 @@ public class ConsumerConfiguration { } } } - } catch (Exception e) { - throw new MessagingException("Consuming from Kafka failed", e); + } + catch (Exception e) { + if (!this.stopped) { + throw new MessagingException("Consuming from Kafka failed", e); + } + else { + LOGGER.warn("Consuming from Kafka failed", e); + } } if (messages.isEmpty()) { @@ -221,4 +265,13 @@ public class ConsumerConfiguration { } return consumerConnector; } + + public void shutdown() { + this.stopped = true; + if (!this.executorExplicitlySet) { + this.executorService.shutdownNow(); + } + getConsumerConnector().shutdown(); + } + } diff --git a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaConsumerContext.java b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaConsumerContext.java index 650717c3bc..197e5e0005 100644 --- a/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaConsumerContext.java +++ b/spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaConsumerContext.java @@ -82,7 +82,7 @@ public class KafkaConsumerContext implements DisposableBean { @Override public void destroy() throws Exception { for (ConsumerConfiguration config : this.consumerConfigurations.values()) { - config.getConsumerConnector().shutdown(); + config.shutdown(); } } diff --git a/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd b/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd index cc89f4d470..8d18422366 100644 --- a/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd +++ b/spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd @@ -311,6 +311,21 @@ + + + + + A java.util.concurrent.Executor bean reference. + Typically + org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor. + Used to execute iterator task on KafkaStreams. + + + + + + + @@ -328,9 +343,14 @@ - - Kafka Server Bean Name - + + + Kafka Server Bean Name + + + + +