spring-projectsGH-1880: SpEL Improvements for Property Overrides

Resolves spring-projects#1880

SpEL can resolve to `String`, `String[]` or `Collection<String>`.

Author: garyrussell

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

@@ -247,6 +247,10 @@
 	 * <li>{@code key value}</li>
 	 * </ul>
 	 * {@code group.id} and {@code client.id} are ignored.
+	 * <p>SpEL {@code #{...}} and property place holders {@code ${...}} are supported.
+	 * SpEL expressions must resolve to a {@link String}, a @{link String[]} or a
+	 * {@code Collection<String>} where each member of the array or collection is a
+	 * property name + value with the above formats.
 	 * @return the properties.
 	 * @since 2.2.4
 	 * @see org.apache.kafka.clients.consumer.ConsumerConfig

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

@@ -676,24 +676,46 @@ protected String noBeanFoundMessage(Object target, String listenerBeanName, Stri
 				+ requestedBeanName + "' was found in the application context";
 	}
 
+	@SuppressWarnings("unchecked")
 	private void resolveKafkaProperties(MethodKafkaListenerEndpoint<?, ?> endpoint, String[] propertyStrings) {
 		if (propertyStrings.length > 0) {
 			Properties properties = new Properties();
 			for (String property : propertyStrings) {
-				String value = resolveExpressionAsString(property, "property");
-				if (value != null) {
-					try {
-						properties.load(new StringReader(value));
+				Object value = resolveExpression(property);
+				if (value instanceof String) {
+					loadProperty(properties, property, value);
+				}
+				else if (value instanceof String[]) {
+					for (String prop : (String[]) value) {
+						loadProperty(properties, prop, prop);
 					}
-					catch (IOException e) {
-						this.logger.error(e, () -> "Failed to load property " + property + ", continuing...");
+				}
+				else if (value instanceof Collection) {
+					Collection<?> values = (Collection<?>) value;
+					if (values.size() > 0 && values.iterator().next() instanceof String) {
+						for (String prop : (Collection<String>) value) {
+							loadProperty(properties, prop, prop);
+						}
 					}
 				}
+				else {
+					throw new IllegalStateException("'properties' must resolve to a String, a String[] or "
+							+ "Collection<String>");
+				}
 			}
 			endpoint.setConsumerProperties(properties);
 		}
 	}
 
+	private void loadProperty(Properties properties, String property, Object value) {
+		try {
+			properties.load(new StringReader((String) value));
+		}
+		catch (IOException e) {
+			this.logger.error(e, () -> "Failed to load property " + property + ", continuing...");
+		}
+	}
+
 	private String getEndpointId(KafkaListener kafkaListener) {
 		if (StringUtils.hasText(kafkaListener.id())) {
 			return resolveExpressionAsString(kafkaListener.id(), "id");

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

@@ -144,6 +144,7 @@
 import org.springframework.retry.support.RetryTemplate;
 import org.springframework.stereotype.Component;
 import org.springframework.test.annotation.DirtiesContext;
+import org.springframework.test.context.TestPropertySource;
 import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
 import org.springframework.transaction.PlatformTransactionManager;
 import org.springframework.transaction.annotation.EnableTransactionManagement;
@@ -177,6 +178,7 @@
 		"annotated29", "annotated30", "annotated30reply", "annotated31", "annotated32", "annotated33",
 		"annotated34", "annotated35", "annotated36", "annotated37", "foo", "manualStart", "seekOnIdle",
 		"annotated38", "annotated38reply", "annotated39", "annotated40", "annotated41" })
+@TestPropertySource(properties = "spel.props=fetch.min.bytes=420000,max.poll.records=10")
 public class EnableKafkaIntegrationTests {
 
 	private static final String DEFAULT_TEST_GROUP_ID = "testAnnot";
@@ -350,6 +352,10 @@ public void testSimple() throws Exception {
 				.isEqualTo("fiz");
 		assertThat(KafkaTestUtils.getPropertyValue(fizContainer, "listenerConsumer.consumer.clientId"))
 				.isEqualTo("clientIdViaAnnotation-0");
+		assertThat(KafkaTestUtils.getPropertyValue(fizContainer, "listenerConsumer.consumer.fetcher.maxPollRecords"))
+				.isEqualTo(10);
+		assertThat(KafkaTestUtils.getPropertyValue(fizContainer, "listenerConsumer.consumer.fetcher.minBytes"))
+				.isEqualTo(420000);
 
 		MessageListenerContainer rebalanceConcurrentContainer = registry.getListenerContainer("rebalanceListener");
 		assertThat(rebalanceConcurrentContainer).isNotNull();
@@ -488,6 +494,10 @@ public void testJson() throws Exception {
 		assertThat(KafkaTestUtils.getPropertyValue(buzContainer,
 						"listenerConsumer.consumer.groupId", Optional.class).get())
 				.isEqualTo("buz.explicitGroupId");
+		assertThat(KafkaTestUtils.getPropertyValue(buzContainer, "listenerConsumer.consumer.fetcher.maxPollRecords"))
+				.isEqualTo(5);
+		assertThat(KafkaTestUtils.getPropertyValue(buzContainer, "listenerConsumer.consumer.fetcher.minBytes"))
+				.isEqualTo(123456);
 	}
 
 	@Test
@@ -1649,6 +1659,11 @@ protected Object convertToInternal(Object payload, @Nullable MessageHeaders head
 			};
 		}
 
+		@Bean
+		List<String> buzProps() {
+			return List.of("max.poll.records: 5", "fetch.min.bytes: 123456");
+		}
+
 	}
 
 	@Component
@@ -1845,14 +1860,14 @@ public void eventHandler(ListenerContainerNoLongerIdleEvent event) {
 				@TopicPartition(topic = "annotated6", partitions = "0",
 						partitionOffsets = @PartitionOffset(partition = "${xxx:1}", initialOffset = "${yyy:0}",
 								relativeToCurrent = "${zzz:true}"))
-		}, clientIdPrefix = "${foo.xxx:clientIdViaAnnotation}")
+		}, clientIdPrefix = "${foo.xxx:clientIdViaAnnotation}", properties = "#{'${spel.props}'.split(',')}")
 		public void listen5(ConsumerRecord<?, ?> record) {
 			this.capturedRecord = record;
 			this.latch5.countDown();
 		}
 
 		@KafkaListener(id = "buz", topics = "annotated10", containerFactory = "kafkaJsonListenerContainerFactory",
-				groupId = "buz.explicitGroupId")
+				groupId = "buz.explicitGroupId", properties = "#{@buzProps}")
 		public void listen6(Foo foo) {
 			this.foo = foo;
 			this.latch6.countDown();