GH-8720: Check MQTT topics if not empty strings

Fixes #8720

Validate MQTT topics for empty strings in the channel adapters configuration

Use plural names for varargs params

**Cherry-pick to `6.1.x` & `6.0.x`**

# Conflicts:
#	spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java
#	spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java
#	spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java

Author: artembilan

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/AbstractMqttMessageDrivenChannelAdapter.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2022 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -78,12 +78,22 @@ public AbstractMqttMessageDrivenChannelAdapter(@Nullable String url, String clie
 		Assert.noNullElements(topic, "'topics' cannot have null elements");
 		this.url = url;
 		this.clientId = clientId;
+		validateTopics(topic);
 		this.topics = new LinkedHashSet<>();
 		for (String t : topic) {
 			this.topics.add(new Topic(t, 1));
 		}
 	}
 
+	private static void validateTopics(String[] topics) {
+		Assert.notNull(topics, "'topics' cannot be null");
+		Assert.noNullElements(topics, "'topics' cannot have null elements");
+
+		for (String topic : topics) {
+			Assert.hasText(topic, "The topic to subscribe cannot be empty string");
+		}
+	}
+
 	public void setConverter(MqttMessageConverter converter) {
 		Assert.notNull(converter, "'converter' cannot be null");
 		this.converter = converter;
@@ -207,6 +217,7 @@ protected long getCompletionTimeout() {
 	 */
 	@ManagedOperation
 	public void addTopic(String topic, int qos) {
+		validateTopics(new String[] {topic});
 		this.topicLock.lock();
 		try {
 			Topic topik = new Topic(topic, qos);
@@ -223,16 +234,16 @@ public void addTopic(String topic, int qos) {
 
 	/**
 	 * Add a topic (or topics) to the subscribed list (qos=1).
-	 * @param topic The topics.
-	 * @throws MessagingException if the topic is already in the list.
+	 * @param topics The topics.
+	 * @throws MessagingException if the topics is already in the list.
 	 * @since 4.1
 	 */
 	@ManagedOperation
-	public void addTopic(String... topic) {
-		Assert.notNull(topic, "'topic' cannot be null");
+	public void addTopic(String... topics) {
+		validateTopics(topics);
 		this.topicLock.lock();
 		try {
-			for (String t : topic) {
+			for (String t : topics) {
 				addTopic(t, 1);
 			}
 		}
@@ -243,25 +254,24 @@ public void addTopic(String... topic) {
 
 	/**
 	 * Add topics to the subscribed list.
-	 * @param topic The topics.
+	 * @param topics The topics.
 	 * @param qos The qos for each topic.
-	 * @throws MessagingException if a topic is already in the list.
+	 * @throws MessagingException if a topics is already in the list.
 	 * @since 4.1
 	 */
 	@ManagedOperation
-	public void addTopics(String[] topic, int[] qos) {
-		Assert.notNull(topic, "'topic' cannot be null.");
-		Assert.noNullElements(topic, "'topic' cannot contain any null elements.");
-		Assert.isTrue(topic.length == qos.length, "topic and qos arrays must the be the same length.");
+	public void addTopics(String[] topics, int[] qos) {
+		validateTopics(topics);
+		Assert.isTrue(topics.length == qos.length, "topics and qos arrays must the be the same length.");
 		this.topicLock.lock();
 		try {
-			for (String topik : topic) {
-				if (this.topics.contains(new Topic(topik, 0))) {
-					throw new MessagingException("Topic '" + topik + "' is already subscribed.");
+			for (String topic : topics) {
+				if (this.topics.contains(new Topic(topic, 0))) {
+					throw new MessagingException("Topic '" + topic + "' is already subscribed.");
 				}
 			}
-			for (int i = 0; i < topic.length; i++) {
-				addTopic(topic[i], qos[i]);
+			for (int i = 0; i < topics.length; i++) {
+				addTopic(topics[i], qos[i]);
 			}
 		}
 		finally {

spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/outbound/AbstractMqttMessageHandler.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2021 the original author or authors.
+ * Copyright 2002-2023 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.
@@ -18,6 +18,7 @@
 
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import org.springframework.beans.factory.BeanFactory;
 import org.springframework.beans.factory.BeanFactoryAware;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
@@ -107,6 +108,7 @@ protected ApplicationEventPublisher getApplicationEventPublisher() {
 	 * @param defaultTopic the default topic.
 	 */
 	public void setDefaultTopic(String defaultTopic) {
+		Assert.hasText(defaultTopic, "'defaultTopic' must not be empty");
 		this.defaultTopic = defaultTopic;
 	}
 
@@ -295,14 +297,17 @@ protected long getDisconnectCompletionTimeout() {
 	@Override
 	protected void onInit() {
 		super.onInit();
-		if (this.topicProcessor instanceof BeanFactoryAware && getBeanFactory() != null) {
-			((BeanFactoryAware) this.topicProcessor).setBeanFactory(getBeanFactory());
-		}
-		if (this.qosProcessor instanceof BeanFactoryAware && getBeanFactory() != null) {
-			((BeanFactoryAware) this.qosProcessor).setBeanFactory(getBeanFactory());
-		}
-		if (this.retainedProcessor instanceof BeanFactoryAware && getBeanFactory() != null) {
-			((BeanFactoryAware) this.retainedProcessor).setBeanFactory(getBeanFactory());
+		BeanFactory beanFactory = getBeanFactory();
+		if (beanFactory != null) {
+			if (this.topicProcessor instanceof BeanFactoryAware) {
+				((BeanFactoryAware) this.topicProcessor).setBeanFactory(beanFactory);
+			}
+			if (this.qosProcessor instanceof BeanFactoryAware) {
+				((BeanFactoryAware) this.qosProcessor).setBeanFactory(beanFactory);
+			}
+			if (this.retainedProcessor instanceof BeanFactoryAware) {
+				((BeanFactoryAware) this.retainedProcessor).setBeanFactory(beanFactory);
+			}
 		}
 	}
 
@@ -333,11 +338,13 @@ public boolean isRunning() {
 	protected void handleMessageInternal(Message<?> message) {
 		Object mqttMessage = this.converter.fromMessage(message, Object.class);
 		String topic = this.topicProcessor.processMessage(message);
-		if (topic == null && this.defaultTopic == null) {
-			throw new IllegalStateException(
-					"No topic could be determined from the message and no default topic defined");
+		if (topic == null) {
+			topic = this.defaultTopic;
 		}
-		publish(topic == null ? this.defaultTopic : topic, mqttMessage, message);
+
+		Assert.state(topic != null, "No topic could be determined from the message and no default topic defined");
+
+		publish(topic, mqttMessage, message);
 	}
 
 	protected abstract void publish(String topic, Object mqttMessage, Message<?> message);

spring-integration-mqtt/src/test/java/org/springframework/integration/mqtt/MqttAdapterTests.java

@@ -62,6 +62,7 @@
 import org.springframework.integration.channel.QueueChannel;
 import org.springframework.integration.handler.MessageProcessor;
 import org.springframework.integration.mqtt.core.DefaultMqttPahoClientFactory;
+import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
 import org.springframework.integration.mqtt.event.MqttConnectionFailedEvent;
 import org.springframework.integration.mqtt.event.MqttIntegrationEvent;
 import org.springframework.integration.mqtt.event.MqttSubscribedEvent;
@@ -83,6 +84,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatExceptionOfType;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.assertj.core.api.Assertions.fail;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
@@ -468,7 +470,7 @@ public void testSubscribeFailure() throws Exception {
 		willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any());
 
 		IMqttToken token = mock(IMqttToken.class);
-		given(token.getGrantedQos()).willReturn(new int[]{ 0x80 });
+		given(token.getGrantedQos()).willReturn(new int[] {0x80});
 		willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), isNull(), isNull(), any());
 
 		MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory,
@@ -516,7 +518,7 @@ public void testDifferentQos() throws Exception {
 		willReturn(alwaysComplete).given(aClient).connect(any(MqttConnectOptions.class), any(), any());
 
 		IMqttToken token = mock(IMqttToken.class);
-		given(token.getGrantedQos()).willReturn(new int[]{ 2, 0 });
+		given(token.getGrantedQos()).willReturn(new int[] {2, 0});
 		willReturn(token).given(aClient).subscribe(any(String[].class), any(int[].class), isNull(), isNull(), any());
 
 		MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter("foo", "bar", factory,
@@ -543,6 +545,19 @@ public void testDifferentQos() throws Exception {
 		verify(client).disconnectForcibly(5_000L);
 	}
 
+	@Test
+	public void emptyTopicNotAllowed() {
+		assertThatIllegalArgumentException()
+				.isThrownBy(() ->
+						new MqttPahoMessageDrivenChannelAdapter("client_id", mock(MqttPahoClientFactory.class), ""))
+				.withMessage("The topic to subscribe cannot be empty string");
+
+		var adapter = new MqttPahoMessageDrivenChannelAdapter("client_id", mock(MqttPahoClientFactory.class), "topic1");
+		assertThatIllegalArgumentException()
+				.isThrownBy(() -> adapter.addTopic(""))
+				.withMessage("The topic to subscribe cannot be empty string");
+	}
+
 	@Test
 	public void testNoNPEOnReconnectAndStopRaceCondition() throws Exception {
 		final IMqttClient client = mock(IMqttClient.class);
@@ -597,7 +612,7 @@ public IMqttClient getClientInstance(String uri, String clientId) throws MqttExc
 
 		};
 		MqttConnectOptions connectOptions = new MqttConnectOptions();
-		connectOptions.setServerURIs(new String[]{ "tcp://localhost:1883" });
+		connectOptions.setServerURIs(new String[] {"tcp://localhost:1883"});
 		if (cleanSession != null) {
 			connectOptions.setCleanSession(cleanSession);
 		}
@@ -621,7 +636,7 @@ public IMqttAsyncClient getAsyncClientInstance(String uri, String clientId) {
 
 		};
 		MqttConnectOptions connectOptions = new MqttConnectOptions();
-		connectOptions.setServerURIs(new String[]{ "tcp://localhost:1883" });
+		connectOptions.setServerURIs(new String[] {"tcp://localhost:1883"});
 		factory.setConnectionOptions(connectOptions);
 		MqttPahoMessageHandler adapter = new MqttPahoMessageHandler("client", factory);
 		adapter.setDefaultTopic("foo");