GH-756: Add concurrency, autoStartup to @kl

Resolves #756

Override the defaults configured on the container factory.

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListener.java

@@ -186,4 +186,19 @@
 	 */
 	String beanRef() default "__listener";
 
+	/**
+	 * Override the container factory's {@code concurrency} setting for this listener.
+	 * May be a property placeholder or SpEL expression that evaluates to an integer.
+	 * @return the concurrency.
+	 * @since 2.2
+	 */
+	String concurrency() default "";
+
+	/**
+	 * Set to true or false, to override the default setting in the container factory.
+	 * @return true to auto start, false to not auto start.
+	 * @since 2.2
+	 */
+	String autoStartup() default "";
+
 }

spring-kafka/src/main/java/org/springframework/kafka/annotation/KafkaListenerAnnotationBeanPostProcessor.java

@@ -418,6 +418,14 @@ protected void processListener(MethodKafkaListenerEndpoint<?, ?> endpoint, Kafka
 				endpoint.setGroup((String) resolvedGroup);
 			}
 		}
+		String concurrency = kafkaListener.concurrency();
+		if (StringUtils.hasText(concurrency)) {
+			endpoint.setConcurrency(resolveExpressionAsInteger(concurrency, "concurrency"));
+		}
+		String autoStartup = kafkaListener.autoStartup();
+		if (StringUtils.hasText(autoStartup)) {
+			endpoint.setAutoStartup(resolveExpressionAsBoolean(autoStartup, "autoStartup"));
+		}
 
 		KafkaListenerContainerFactory<?> factory = null;
 		String containerFactoryBeanName = resolve(kafkaListener.containerFactory());
@@ -640,6 +648,35 @@ private String resolveExpressionAsString(String value, String attribute) {
 		}
 	}
 
+	private int resolveExpressionAsInteger(String value, String attribute) {
+		Object resolved = resolveExpression(value);
+		if (resolved instanceof String) {
+			return Integer.parseInt((String) resolved);
+		}
+		else if (resolved instanceof Number) {
+			return ((Number) resolved).intValue();
+		}
+		else {
+			throw new IllegalStateException("The [" + attribute + "] must resolve to a String. "
+					+ "Resolved to [" + resolved.getClass() + "] for [" + value + "]");
+		}
+	}
+
+	private boolean resolveExpressionAsBoolean(String value, String attribute) {
+		Object resolved = resolveExpression(value);
+		if (resolved instanceof Boolean) {
+			return (Boolean) resolved;
+		}
+		else if (resolved instanceof String) {
+			final String s = (String) resolved;
+			return Boolean.parseBoolean(s);
+		}
+		else {
+			throw new IllegalStateException("The [" + attribute + "] must resolve to a Boolean. "
+					+ "Resolved to [" + resolved.getClass() + "] for [" + value + "]");
+		}
+	}
+
 	private Object resolveExpression(String value) {
 		String resolvedValue = resolve(value);
 

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

@@ -247,7 +247,6 @@ public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 		if (endpoint.getId() != null) {
 			instance.setBeanName(endpoint.getId());
 		}
-
 		if (endpoint instanceof AbstractKafkaListenerEndpoint) {
 			AbstractKafkaListenerEndpoint<K, V> aklEndpoint = (AbstractKafkaListenerEndpoint<K, V>) endpoint;
 			if (this.recordFilterStrategy != null) {
@@ -274,7 +273,7 @@ public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 		}
 
 		endpoint.setupListenerContainer(instance, this.messageConverter);
-		initializeContainer(instance);
+		initializeContainer(instance, endpoint);
 		instance.getContainerProperties().setGroupId(endpoint.getGroupId());
 		instance.getContainerProperties().setClientId(endpoint.getClientIdPrefix());
 
@@ -293,8 +292,9 @@ public C createListenerContainer(KafkaListenerEndpoint endpoint) {
 	 * <p>Subclasses can inherit from this method to apply extra
 	 * configuration if necessary.
 	 * @param instance the container instance to configure.
+	 * @param endpoint the endpoint.
 	 */
-	protected void initializeContainer(C instance) {
+	protected void initializeContainer(C instance, KafkaListenerEndpoint endpoint) {
 		ContainerProperties properties = instance.getContainerProperties();
 		BeanUtils.copyProperties(this.containerProperties, properties, "topics", "topicPartitions", "topicPattern",
 				"messageListener", "ackCount", "ackTime");
@@ -310,7 +310,10 @@ protected void initializeContainer(C instance) {
 		if (this.errorHandler != null) {
 			instance.setGenericErrorHandler(this.errorHandler);
 		}
-		if (this.autoStartup != null) {
+		if (endpoint.getAutoStartup() != null) {
+			instance.setAutoStartup(endpoint.getAutoStartup());
+		}
+		else if (this.autoStartup != null) {
 			instance.setAutoStartup(this.autoStartup);
 		}
 		if (this.phase != null) {
@@ -323,43 +326,46 @@ protected void initializeContainer(C instance) {
 
 	@Override
 	public C createContainer(final Collection<TopicPartitionInitialOffset> topicPartitions) {
-		C container = createContainerInstance(new KafkaListenerEndpointAdapter() {
+		KafkaListenerEndpoint endpoint = new KafkaListenerEndpointAdapter() {
 
 					@Override
 					public Collection<TopicPartitionInitialOffset> getTopicPartitions() {
 						return topicPartitions;
 					}
 
-				});
-		initializeContainer(container);
+				};
+		C container = createContainerInstance(endpoint);
+		initializeContainer(container, endpoint);
 		return container;
 	}
 
 	@Override
 	public C createContainer(final String... topics) {
-		C container = createContainerInstance(new KafkaListenerEndpointAdapter() {
+		KafkaListenerEndpoint endpoint = new KafkaListenerEndpointAdapter() {
 
 					@Override
 					public Collection<String> getTopics() {
 						return Arrays.asList(topics);
 					}
 
-				});
-		initializeContainer(container);
+				};
+		C container = createContainerInstance(endpoint);
+		initializeContainer(container, endpoint);
 		return container;
 	}
 
 	@Override
 	public C createContainer(final Pattern topicPattern) {
-		C container = createContainerInstance(new KafkaListenerEndpointAdapter() {
+		KafkaListenerEndpoint endpoint = new KafkaListenerEndpointAdapter() {
 
 					@Override
 					public Pattern getTopicPattern() {
 						return topicPattern;
 					}
 
-				});
-		initializeContainer(container);
+				};
+		C container = createContainerInstance(endpoint);
+		initializeContainer(container, endpoint);
 		return container;
 	}
 

spring-kafka/src/main/java/org/springframework/kafka/config/AbstractKafkaListenerEndpoint.java

@@ -101,6 +101,10 @@ public abstract class AbstractKafkaListenerEndpoint<K, V>
 
 	private String clientIdPrefix;
 
+	private Integer concurrency;
+
+	private Boolean autoStartup;
+
 	@Override
 	public void setBeanFactory(BeanFactory beanFactory) throws BeansException {
 		this.beanFactory = beanFactory;
@@ -342,6 +346,34 @@ public void setClientIdPrefix(String clientIdPrefix) {
 		this.clientIdPrefix = clientIdPrefix;
 	}
 
+	@Override
+	public Integer getConcurrency() {
+		return this.concurrency;
+	}
+
+	/**
+	 * Set the concurrency for this endpoint's container.
+	 * @param concurrency the concurrency.
+	 * @since 2.2
+	 */
+	public void setConcurrency(int concurrency) {
+		this.concurrency = concurrency;
+	}
+
+	@Override
+	public Boolean getAutoStartup() {
+		return this.autoStartup;
+	}
+
+	/**
+	 * Set the autoStartup for this endpoint's container.
+	 * @param autoStartup the autoStartup.
+	 * @since 2.2
+	 */
+	public void setAutoStartup(Boolean autoStartup) {
+		this.autoStartup = autoStartup;
+	}
+
 	@Override
 	public void afterPropertiesSet() {
 		boolean topicsEmpty = getTopics().isEmpty();

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

@@ -79,9 +79,12 @@ protected ConcurrentMessageListenerContainer<K, V> createContainerInstance(Kafka
 	}
 
 	@Override
-	protected void initializeContainer(ConcurrentMessageListenerContainer<K, V> instance) {
-		super.initializeContainer(instance);
-		if (this.concurrency != null) {
+	protected void initializeContainer(ConcurrentMessageListenerContainer<K, V> instance, KafkaListenerEndpoint endpoint) {
+		super.initializeContainer(instance, endpoint);
+		if (endpoint.getConcurrency() != null) {
+			instance.setConcurrency(endpoint.getConcurrency());
+		}
+		else if (this.concurrency != null) {
 			instance.setConcurrency(this.concurrency);
 		}
 	}

spring-kafka/src/main/java/org/springframework/kafka/config/KafkaListenerEndpoint.java

@@ -83,6 +83,20 @@ public interface KafkaListenerEndpoint {
 	 */
 	String getClientIdPrefix();
 
+	/**
+	 * Return the concurrency for this endpoint's container.
+	 * @return the concurrency.
+	 * @since 2.2
+	 */
+	Integer getConcurrency();
+
+	/**
+	 * Return the autoStartup for this endpoint's container.
+	 * @return the autoStartup.
+	 * @since 2.2
+	 */
+	Boolean getAutoStartup();
+
 	/**
 	 * Setup the specified message listener container with the model
 	 * defined by this endpoint.

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

@@ -72,6 +72,16 @@ public String getClientIdPrefix() {
 		return null;
 	}
 
+	@Override
+	public Integer getConcurrency() {
+		return null;
+	}
+
+	@Override
+	public Boolean getAutoStartup() {
+		return null;
+	}
+
 	@Override
 	public void setupListenerContainer(MessageListenerContainer listenerContainer,
 			MessageConverter messageConverter) {

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -255,6 +255,13 @@ public void testSimple() throws Exception {
 		assertThat(KafkaTestUtils.getPropertyValue(fizContainer, "listenerConsumer.consumer.clientId"))
 				.isEqualTo("clientIdViaAnnotation-0");
 
+		MessageListenerContainer rebalanceConcurrentContainer = registry.getListenerContainer("rebalanceListener");
+		assertThat(rebalanceConcurrentContainer).isNotNull();
+		assertThat(rebalanceConcurrentContainer.isAutoStartup()).isFalse();
+		assertThat(KafkaTestUtils.getPropertyValue(rebalanceConcurrentContainer, "concurrency", Integer.class))
+				.isEqualTo(3);
+		rebalanceConcurrentContainer.start();
+
 		template.send("annotated11", 0, "foo");
 		assertThat(this.listener.latch7.await(60, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.consumerRef.get()).isNotNull();
@@ -266,8 +273,6 @@ public void testSimple() throws Exception {
 		assertThat(this.listener.latch7a.await(60, TimeUnit.SECONDS)).isTrue();
 		assertThat(this.listener.latch7String).isNull();
 
-		MessageListenerContainer rebalanceConcurrentContainer = registry.getListenerContainer("rebalanceListener");
-		assertThat(rebalanceConcurrentContainer).isNotNull();
 		MessageListenerContainer rebalanceContainer = (MessageListenerContainer) KafkaTestUtils
 				.getPropertyValue(rebalanceConcurrentContainer, "containers", List.class).get(0);
 		assertThat(KafkaTestUtils.getPropertyValue(rebalanceContainer, "listenerConsumer.consumer.coordinator.groupId"))
@@ -278,7 +283,6 @@ public void testSimple() throws Exception {
 				clientId)
 				.startsWith("consumer-");
 		assertThat(clientId.indexOf('-')).isEqualTo(clientId.lastIndexOf('-'));
-
 	}
 
 	@Test
@@ -847,6 +851,7 @@ public KafkaListenerContainerFactory<?> batchManualFactory2() {
 					new ConcurrentKafkaListenerContainerFactory<>();
 			ContainerProperties props = factory.getContainerProperties();
 			factory.setConsumerFactory(consumerFactory());
+			factory.setAutoStartup(true);
 			props.setConsumerRebalanceListener(consumerRebalanceListener(consumerRef()));
 			return factory;
 		}
@@ -1268,7 +1273,8 @@ public void jsonHeaders(Bar foo) { // should be mapped to Foo via Headers
 		}
 
 		@KafkaListener(id = "rebalanceListener", topics = "annotated11", idIsGroup = false,
-				containerFactory = "kafkaRebalanceListenerContainerFactory")
+				containerFactory = "kafkaRebalanceListenerContainerFactory", autoStartup = "${foobarbaz:false}",
+				concurrency = "${fixbux:3}")
 		public void listen7(@Payload(required = false) String foo) {
 			this.latch7String = foo;
 			this.latch7.countDown();