Resolve @RedisListener consumes placeholders

src/main/java/org/springframework/data/redis/annotation/RedisListenerAnnotationBeanPostProcessor.java

@@ -69,6 +69,7 @@
  * @author Ilyass Bougati
  * @author Mark Paluch
  * @author Christoph Strobl
+ * @author Dongliang Xie
  * @since 4.1
  * @see RedisListener
  */
@@ -249,7 +250,7 @@ public MethodRedisListenerEndpoint createEndpoint(RedisListener redisListener, M
 		endpoint.setMessageHandlerMethodFactory(this.messageHandlerMethodFactory);
 		endpoint.setId(getEndpointId(redisListener));
 		endpoint.setTopic(redisListener.topic());
-		endpoint.setConsumes(redisListener.consumes());
+		endpoint.setConsumes(resolve(redisListener.consumes()));
 
 		return endpoint;
 	}

src/test/java/org/springframework/data/redis/annotation/RedisListenerAnnotationBeanPostProcessorIntegrationTests.java

@@ -19,27 +19,33 @@
 import static org.mockito.ArgumentMatchers.*;
 import static org.mockito.Mockito.*;
 
+import java.util.Map;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 
 import org.springframework.beans.factory.BeanCreationException;
 import org.springframework.beans.factory.NoSuchBeanDefinitionException;
 import org.springframework.context.ApplicationContext;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.MapPropertySource;
+import org.springframework.data.redis.connection.MessageListener;
 import org.springframework.data.redis.config.RedisListenerConfigUtils;
 import org.springframework.data.redis.config.RedisListenerEndpointRegistry;
 import org.springframework.data.redis.listener.RedisMessageListenerContainer;
+import org.springframework.data.redis.listener.StringMessage;
 import org.springframework.data.redis.listener.Topic;
 
 /**
  * Integration test for {@link EnableRedisListeners} and {@link RedisListener}
  *
  * @author Ilyass Bougati
  * @author Mark Paluch
+ * @author Dongliang Xie
  */
 class RedisListenerAnnotationBeanPostProcessorIntegrationTests {
 
@@ -62,6 +68,30 @@ void registersListenerWithDefaultContainer() {
 		assertThat(registryRef.get().isRunning()).isFalse();
 	}
 
+	@Test
+	void resolvesListenerConsumesPlaceholder() {
+
+		try (AnnotationConfigApplicationContext context = new AnnotationConfigApplicationContext()) {
+			context.getEnvironment().getPropertySources()
+					.addFirst(new MapPropertySource("redis-listener-test",
+							Map.of("app.redis.content-type", "application/json")));
+			context.register(DefaultConfig.class, ConsumesPlaceholderService.class);
+			context.refresh();
+
+			RedisMessageListenerContainer container = context.getBean("redisMessageListenerContainer",
+					RedisMessageListenerContainer.class);
+			ArgumentCaptor<MessageListener> listenerCaptor = ArgumentCaptor.forClass(MessageListener.class);
+
+			verify(container).addMessageListener(listenerCaptor.capture(), any(Topic.class));
+
+			listenerCaptor.getValue().onMessage(
+					new StringMessage("test-topic", "{\"firstname\":\"Walter\",\"lastname\":\"White\"}"), null);
+
+			ConsumesPlaceholderService service = context.getBean(ConsumesPlaceholderService.class);
+			assertThat(service.person.get()).isEqualTo(new Person("Walter", "White"));
+		}
+	}
+
 	@Test // GH-3340
 	void registersListenerWithNamedContainer() {
 
@@ -140,6 +170,21 @@ public void handle(String msg) {}
 
 	}
 
+	static class ConsumesPlaceholderService {
+
+		final AtomicReference<Person> person = new AtomicReference<>();
+
+		@RedisListener(topic = "test-topic", consumes = "${app.redis.content-type}")
+		public void handle(Person person) {
+			this.person.set(person);
+		}
+
+	}
+
+	record Person(String firstname, String lastname) {
+
+	}
+
 	static class UnnamedContainerService {
 
 		@RedisListener(topic = "test-topic", container = "")