GH-1770: Fix ClientId for @KafkaListener

Resolves #1770

The clientId property, if set on the container factory, is overwritten
by an empty String if not defined on the annotation.

Check for an empty String (groupId too).

**cherry-pick to 2.6.x**

* Use simpler JavaUtils method.

Author: garyrussell

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

@@ -411,8 +411,8 @@ else if (this.autoStartup != null) {
 				.acceptIfNotNull(this.phase, instance::setPhase)
 				.acceptIfNotNull(this.applicationContext, instance::setApplicationContext)
 				.acceptIfNotNull(this.applicationEventPublisher, instance::setApplicationEventPublisher)
-				.acceptIfNotNull(endpoint.getGroupId(), instance.getContainerProperties()::setGroupId)
-				.acceptIfNotNull(endpoint.getClientIdPrefix(), instance.getContainerProperties()::setClientId)
+				.acceptIfHasText(endpoint.getGroupId(), instance.getContainerProperties()::setGroupId)
+				.acceptIfHasText(endpoint.getClientIdPrefix(), instance.getContainerProperties()::setClientId)
 				.acceptIfNotNull(endpoint.getConsumerProperties(),
 						instance.getContainerProperties()::setKafkaConsumerProperties);
 	}

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

@@ -1,5 +1,5 @@
 /*
- * Copyright 2018-2019 the original author or authors.
+ * Copyright 2018-2021 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.
@@ -23,9 +23,14 @@
 
 import org.junit.jupiter.api.Test;
 
+import org.springframework.kafka.config.AbstractKafkaListenerEndpoint;
 import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
 import org.springframework.kafka.core.ConsumerFactory;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
+import org.springframework.kafka.listener.MessageListenerContainer;
+import org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter;
+import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter;
+import org.springframework.kafka.support.converter.MessageConverter;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 
 /**
@@ -36,7 +41,7 @@
 public class ContainerFactoryTests {
 
 	@Test
-	public void testConfigContainer() {
+	void testConfigContainer() {
 		ConcurrentKafkaListenerContainerFactory<String, String> factory =
 				new ConcurrentKafkaListenerContainerFactory<>();
 		factory.setAutoStartup(false);
@@ -56,4 +61,33 @@ public void testConfigContainer() {
 		assertThat(customized).isTrue();
 	}
 
+	@SuppressWarnings("unchecked")
+	@Test
+	void clientIdAndGroupIdTransferred() {
+		ConcurrentKafkaListenerContainerFactory<String, String> factory =
+				new ConcurrentKafkaListenerContainerFactory<>();
+		factory.getContainerProperties().setClientId("myClientId");
+		factory.getContainerProperties().setGroupId("myGroup");
+		factory.setConsumerFactory(mock(ConsumerFactory.class));
+		AbstractKafkaListenerEndpoint<String, String> endpoint = new AbstractKafkaListenerEndpoint<String, String>() {
+
+			@Override
+			protected MessagingMessageListenerAdapter<String, String> createMessageListener(
+					MessageListenerContainer container, MessageConverter messageConverter) {
+
+				RecordMessagingMessageListenerAdapter<String, String> adapter =
+						new RecordMessagingMessageListenerAdapter<String, String>(null, null);
+				return adapter;
+			}
+
+		};
+		endpoint.setTopics("test");
+		endpoint.setClientIdPrefix("");
+		endpoint.setGroupId("");
+		ConcurrentMessageListenerContainer<String, String> container = factory.createListenerContainer(
+				endpoint);
+		assertThat(container.getContainerProperties().getClientId()).isEqualTo("myClientId");
+		assertThat(container.getContainerProperties().getGroupId()).isEqualTo("myGroup");
+	}
+
 }