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

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

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.ArrayList;
-import java.util.concurrent.*;
+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<K, V> {
 	private static final Log LOGGER = LogFactory.getLog(ConsumerConfiguration.class);
 
 	private final ConsumerMetadata<K, V> consumerMetadata;
+
 	private final ConsumerConnectionProvider consumerConnectionProvider;
+
 	private final MessageLeftOverTracker<K, V> messageLeftOverTracker;
+
 	private ConsumerConnector consumerConnector;
+
 	private volatile int count = 0;
+
 	private int maxMessages = 1;
+
 	private Collection<List<KafkaStream<K, V>>> consumerMessageStreams;
 
-	private final ExecutorService executorService = Executors.newCachedThreadPool();
+	private ExecutorService executorService = Executors.newCachedThreadPool();
+
+	private boolean executorExplicitlySet;
+
+	private volatile boolean stopped;
 
 	public ConsumerConfiguration(final ConsumerMetadata<K, V> consumerMetadata,
 			final ConsumerConnectionProvider consumerConnectionProvider,
@@ -46,6 +70,19 @@ public class ConsumerConfiguration<K, V> {
 		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<K, V> getConsumerMetadata() {
 		return consumerMetadata;
 	}
@@ -75,7 +112,8 @@ public class ConsumerConfiguration<K, V> {
 									}
 								}
 							}
-						} catch (ConsumerTimeoutException cte) {
+						}
+						catch (ConsumerTimeoutException cte) {
 							LOGGER.debug("Consumer timed out");
 						}
 						return rawMessages;
@@ -93,7 +131,7 @@ public class ConsumerConfiguration<K, V> {
 		messages.putAll(getLeftOverMessageMap());
 
 		try {
-			for (final Future<List<MessageAndMetadata<K, V>>> result : executorService.invokeAll(tasks)) {
+			for (final Future<List<MessageAndMetadata<K, V>>> 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<K, V> {
 					}
 				}
 			}
-		} 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<K, V> {
 		}
 		return consumerConnector;
 	}
+
+	public void shutdown() {
+		this.stopped = true;
+		if (!this.executorExplicitlySet) {
+			this.executorService.shutdownNow();
+		}
+		getConsumerConnector().shutdown();
+	}
+
 }

spring-integration-kafka/src/main/java/org/springframework/integration/kafka/support/KafkaConsumerContext.java

@@ -82,7 +82,7 @@ public class KafkaConsumerContext<K, V> implements DisposableBean {
 	@Override
 	public void destroy() throws Exception {
 		for (ConsumerConfiguration<K, V> config : this.consumerConfigurations.values()) {
-			config.getConsumerConnector().shutdown();
+			config.shutdown();
 		}
 	}
 

spring-integration-kafka/src/main/resources/org/springframework/integration/config/xml/spring-integration-kafka-1.0.xsd

@@ -311,6 +311,21 @@
 											</xsd:appinfo>
 										</xsd:annotation>
 									</xsd:attribute>
+									<xsd:attribute name="executor" type="xsd:string">
+										<xsd:annotation>
+											<xsd:appinfo>
+												<xsd:documentation>
+													A java.util.concurrent.Executor bean reference.
+													Typically
+													org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor.
+													Used to execute iterator task on KafkaStreams.
+												</xsd:documentation>
+												<tool:annotation kind="ref">
+													<tool:expected-type type="java.util.concurrent.Executor"/>
+												</tool:annotation>
+											</xsd:appinfo>
+										</xsd:annotation>
+									</xsd:attribute>
 								</xsd:complexType>
 							</xsd:element>
 						</xsd:choice>
@@ -328,9 +343,14 @@
 			</xsd:attribute>
 			<xsd:attribute name="zookeeper-connect" use="required" type="xsd:string">
 				<xsd:annotation>
-					<xsd:documentation>
+					<xsd:appinfo>
-						Kafka Server Bean Name
+						<xsd:documentation>
-					</xsd:documentation>
+							Kafka Server Bean Name
+						</xsd:documentation>
+						<tool:annotation kind="ref">
+							<tool:expected-type type="org.springframework.integration.kafka.support.ZookeeperConnect"/>
+						</tool:annotation>
+					</xsd:appinfo>
 				</xsd:annotation>
 			</xsd:attribute>
 			<xsd:attribute name="consumer-properties" use="optional" type="xsd:string">