SIKGH-204: Support auto config for all containers

See spring-attic/spring-integration-kafka#204

With this in place, we could pass a container factory into the KMDCA and/or a DSL spec,
along with topic configuration.

This would enable boot properties to be used to configure the container (and any container
that is not used for a `@KafkaListener`.

With the integration DSL, something like

```
   IntegrationFlows.from(Kafka.messageDrivenChannelAdapter(containerFactory, String... topics)
                   .handle(...)
```
where the `ConsumerFactory` and `ContainerProperties` are provided by the factory.

* Javadoc polishing.

* Polishing - pull methods up to abstract factory

* Polishing - add setAutoStartup to container interface

Author: garyrussell

Diff for: spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerContainerFactory.java

@@ -17,6 +17,8 @@
 package org.springframework.kafka.config;
 
 
+import java.util.Arrays;
+import java.util.Collection;
 import java.util.regex.Pattern;
 
 import org.springframework.beans.BeanUtils;
@@ -31,6 +33,7 @@
 import org.springframework.kafka.listener.ErrorHandler;
 import org.springframework.kafka.listener.GenericErrorHandler;
 import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
 import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.retry.RecoveryCallback;
 import org.springframework.retry.support.RetryTemplate;
@@ -241,15 +244,6 @@ public ContainerProperties getContainerProperties() {
 	public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 		C instance = createContainerInstance(endpoint);
 
-		if (this.autoStartup != null) {
-			instance.setAutoStartup(this.autoStartup);
-		}
-		if (this.phase != null) {
-			instance.setPhase(this.phase);
-		}
-		if (this.applicationEventPublisher != null) {
-			instance.setApplicationEventPublisher(this.applicationEventPublisher);
-		}
 		if (endpoint.getId() != null) {
 			instance.setBeanName(endpoint.getId());
 		}
@@ -316,6 +310,57 @@ protected void initializeContainer(C instance) {
 		if (this.errorHandler != null) {
 			instance.setGenericErrorHandler(this.errorHandler);
 		}
+		if (this.autoStartup != null) {
+			instance.setAutoStartup(this.autoStartup);
+		}
+		if (this.phase != null) {
+			instance.setPhase(this.phase);
+		}
+		if (this.applicationEventPublisher != null) {
+			instance.setApplicationEventPublisher(this.applicationEventPublisher);
+		}
+	}
+
+	@Override
+	public C createContainer(final Collection<TopicPartitionInitialOffset> topicPartitions) {
+		C container = createContainerInstance(new KafkaListenerEndpointAdapter() {
+
+					@Override
+					public Collection<TopicPartitionInitialOffset> getTopicPartitions() {
+						return topicPartitions;
+					}
+
+				});
+		initializeContainer(container);
+		return container;
+	}
+
+	@Override
+	public C createContainer(final String... topics) {
+		C container = createContainerInstance(new KafkaListenerEndpointAdapter() {
+
+					@Override
+					public Collection<String> getTopics() {
+						return Arrays.asList(topics);
+					}
+
+				});
+		initializeContainer(container);
+		return container;
+	}
+
+	@Override
+	public C createContainer(final Pattern topicPattern) {
+		C container = createContainerInstance(new KafkaListenerEndpointAdapter() {
+
+					@Override
+					public Pattern getTopicPattern() {
+						return topicPattern;
+					}
+
+				});
+		initializeContainer(container);
+		return container;
 	}
 
 }

Diff for: spring-kafka/src/main/java/org/springframework/kafka/config/ConcurrentKafkaListenerContainerFactory.java

@@ -26,8 +26,11 @@
  * A {@link KafkaListenerContainerFactory} implementation to build a
  * {@link ConcurrentMessageListenerContainer}.
  * <p>
- * This should be the default for most users and a good transition paths
- * for those that are used to build such container definitions manually.
+ * This should be the default for most users and a good transition paths for those that
+ * are used to building such container definitions manually.
+ *
+ * This factory is primarily for building containers for {@code KafkaListener} annotated
+ * methods but can also be used to create any container.
  *
  * @param <K> the key type.
  * @param <V> the value type.
@@ -81,3 +84,4 @@ protected void initializeContainer(ConcurrentMessageListenerContainer<K, V> inst
 	}
 
 }
+

Diff for: spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerContainerFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2016 the original author or authors.
+ * Copyright 2002-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.
@@ -16,15 +16,19 @@
 
 package org.springframework.kafka.config;
 
+import java.util.Collection;
+import java.util.regex.Pattern;
+
 import org.springframework.kafka.listener.MessageListenerContainer;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
 
 /**
- * Factory of {@link MessageListenerContainer} based on a
- * {@link KafkaListenerEndpoint} definition.
+ * Factory for {@link MessageListenerContainer}s.
  *
  * @param <C> the {@link MessageListenerContainer} implementation type.
  *
  * @author Stephane Nicoll
+ * @author Gary Russell
  *
  * @see KafkaListenerEndpoint
  */
@@ -37,4 +41,31 @@ public interface KafkaListenerContainerFactory<C extends MessageListenerContaine
 	 */
 	C createListenerContainer(KafkaListenerEndpoint endpoint);
 
+	/**
+	 * Create and configure a container without a listener; used to create containers that
+	 * are not used for KafkaListener annotations.
+	 * @param topicPartitions the topicPartitions.
+	 * @return the container.
+	 * @since 2.2
+	 */
+	C createContainer(Collection<TopicPartitionInitialOffset> topicPartitions);
+
+	/**
+	 * Create and configure a container without a listener; used to create containers that
+	 * are not used for KafkaListener annotations.
+	 * @param topics the topics.
+	 * @return the container.
+	 * @since 2.2
+	 */
+	C createContainer(String... topics);
+
+	/**
+	 * Create and configure a container without a listener; used to create containers that
+	 * are not used for KafkaListener annotations.
+	 * @param topicPattern the topicPattern.
+	 * @return the container.
+	 * @since 2.2
+	 */
+	C createContainer(Pattern topicPattern);
+
 }

Diff for: spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpointAdapter.java

@@ -0,0 +1,80 @@
+/*
+ * 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.config;
+
+import java.util.Collection;
+import java.util.Collections;
+import java.util.regex.Pattern;
+
+import org.springframework.kafka.listener.MessageListenerContainer;
+import org.springframework.kafka.support.TopicPartitionInitialOffset;
+import org.springframework.kafka.support.converter.MessageConverter;
+
+/**
+ * Adapter to avoid having to implement all methods.
+ *
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+class KafkaListenerEndpointAdapter implements KafkaListenerEndpoint {
+
+	KafkaListenerEndpointAdapter() {
+		super();
+	}
+
+	@Override
+	public String getId() {
+		return null;
+	}
+
+	@Override
+	public String getGroupId() {
+		return null;
+	}
+
+	@Override
+	public String getGroup() {
+		return null;
+	}
+
+	@Override
+	public Collection<String> getTopics() {
+		return Collections.emptyList();
+	}
+
+	@Override
+	public Collection<TopicPartitionInitialOffset> getTopicPartitions() {
+		return Collections.emptyList();
+	}
+
+	@Override
+	public Pattern getTopicPattern() {
+		return null;
+	}
+
+	@Override
+	public String getClientIdPrefix() {
+		return null;
+	}
+
+	@Override
+	public void setupListenerContainer(MessageListenerContainer listenerContainer,
+			MessageConverter messageConverter) {
+	}
+
+}

Diff for: spring-kafka/src/main/java/org/springframework/kafka/listener/MessageListenerContainer.java

@@ -104,4 +104,14 @@ default boolean isContainerPaused() {
 		throw new UnsupportedOperationException("This container doesn't support pause/resume");
 	}
 
+	/**
+	 * Set the autoStartup.
+	 * @param autoStartup the autoStartup to set.
+	 * @since 2.2
+	 * @see SmartLifecycle
+	 */
+	default void setAutoStartup(boolean autoStartup) {
+		// empty
+	}
+
 }

Diff for: spring-kafka/src/test/java/org/springframework/kafka/annotation/ContainerFactoryTests.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.annotation;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.mock;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
+import org.springframework.kafka.core.ConsumerFactory;
+import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
+import org.springframework.kafka.test.utils.KafkaTestUtils;
+
+/**
+ * @author Gary Russell
+ * @since 2.2
+ *
+ */
+public class ContainerFactoryTests {
+
+	@Test
+	public void testConfigContainer() {
+		ConcurrentKafkaListenerContainerFactory<String, String> factory =
+				new ConcurrentKafkaListenerContainerFactory<>();
+		factory.setAutoStartup(false);
+		factory.setConcurrency(22);
+		@SuppressWarnings("unchecked")
+		ConsumerFactory<String, String> cf = mock(ConsumerFactory.class);
+		factory.setConsumerFactory(cf);
+		factory.setPhase(42);
+		factory.getContainerProperties().setAckCount(123);
+		ConcurrentMessageListenerContainer<String, String> container = factory.createContainer("foo");
+		assertThat(container.isAutoStartup()).isFalse();
+		assertThat(container.getPhase()).isEqualTo(42);
+		assertThat(container.getContainerProperties().getAckCount()).isEqualTo(123);
+		assertThat(KafkaTestUtils.getPropertyValue(container, "concurrency", Integer.class)).isEqualTo(22);
+	}
+
+}

Diff for: src/reference/asciidoc/kafka.adoc

@@ -1375,6 +1375,15 @@ You can also perform seek operations from `onIdleContainer()` when an idle conta
 
 To arbitrarily seek at runtime, use the callback reference from the `registerSeekCallback` for the appropriate thread.
 
+[[container-factory]]
+===== Container factory
+
+As discussed in <<kafka-listener-annotation>> a `ConcurrentKafkaListenerContainerFactory` is used to create containers for annotated methods.
+
+Starting with _version 2.2_, the same factory can be used to create any `ConcurrentMessageListenerContainer`.
+This might be useful if you want to create several containers with similar properties, or you wish to use some exernally configured factory, such as the one provided by Spring Boot auto configuration.
+Once the container is created, you can further modify its properties, many of which are set by using `container.getContainerProperties()`.
+
 [[pause-resume]]
 ==== Pausing/Resuming Listener Containers
 

Diff for: src/reference/asciidoc/whats-new.adoc

@@ -1,4 +1,4 @@
-=== What's new in 2.1 Since 2.0
+=== What's new in 2.2 Since 2.1
 
 ==== Kafka Client Version
 
@@ -15,3 +15,8 @@ The enum `AckMode` has been moved from `AbstractMessageListenerContainer` to `Co
 ==== After rollback processing
 
 A new `AfterRollbackProcessor` strategy is provided - see <<after-rollback>> for more information.
+
+==== ConcurrentKafkaListenerContainerFactory changes
+
+The `ConcurrentKafkaListenerContainerFactory` can now be used to create/configure any `ConcurrentMessageListenerContainer`, not just those for `@KafkaListener` annotations.
+See <<container-factory>> for more information.