GH-1485: Option to suppress single client.id suffx

See #1485

Author: garyrussell

spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java

@@ -58,6 +58,8 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
 
 	private int concurrency = 1;
 
+	private boolean alwaysClientIdSuffix = true;
+
 	/**
 	 * Construct an instance with the supplied configuration properties.
 	 * The topic partitions are distributed evenly across the delegate
@@ -86,6 +88,16 @@ public void setConcurrency(int concurrency) {
 		this.concurrency = concurrency;
 	}
 
+	/**
+	 * Set to false to suppress adding a suffix to the child container's client.id when
+	 * the concurrency is only 1.
+	 * @param alwaysClientIdSuffix false to suppress, true (default) to include.
+	 * @since 2.2.14
+	 */
+	public void setAlwaysClientIdSuffix(boolean alwaysClientIdSuffix) {
+		this.alwaysClientIdSuffix = alwaysClientIdSuffix;
+	}
+
 	/**
 	 * Return the list of {@link KafkaMessageListenerContainer}s created by
 	 * this container.
@@ -166,7 +178,7 @@ protected void doStart() {
 				if (getApplicationEventPublisher() != null) {
 					container.setApplicationEventPublisher(getApplicationEventPublisher());
 				}
-				container.setClientIdSuffix("-" + i);
+				container.setClientIdSuffix(this.concurrency > 1 || this.alwaysClientIdSuffix ? "-" + i : "");
 				container.setGenericErrorHandler(getGenericErrorHandler());
 				container.setAfterRollbackProcessor(getAfterRollbackProcessor());
 				container.setRecordInterceptor(getRecordInterceptor());

spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java

@@ -29,6 +29,7 @@
 import java.util.BitSet;
 import java.util.Collection;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -403,6 +404,7 @@ public void testManualCommitSyncExisting() throws Exception {
 			latch.countDown();
 		});
 		containerProps.setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
+		containerProps.setClientId("myClientId");
 
 		ConcurrentMessageListenerContainer<Integer, String> container =
 				new ConcurrentMessageListenerContainer<>(cf, containerProps);
@@ -417,6 +419,9 @@ public void testManualCommitSyncExisting() throws Exception {
 		template.flush();
 		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
 		assertThat(bitSet.cardinality()).isEqualTo(8);
+		Set<String> clientIds = container.metrics().keySet();
+		assertThat(clientIds).hasSize(1);
+		assertThat(clientIds.iterator().next()).isEqualTo("myClientId-0");
 		container.stop();
 		this.logger.info("Stop MANUAL_IMMEDIATE with Existing");
 	}
@@ -433,10 +438,11 @@ public void testPausedStart() throws Exception {
 			ConcurrentMessageListenerContainerTests.this.logger.info("paused start: " + message);
 			latch.countDown();
 		});
-
+		containerProps.setClientId("myClientId");
 		ConcurrentMessageListenerContainer<Integer, String> container =
 				new ConcurrentMessageListenerContainer<>(cf, containerProps);
 		container.setConcurrency(2);
+		container.setAlwaysClientIdSuffix(false);
 		container.setBeanName("testBatch");
 		container.pause();
 		container.start();
@@ -455,6 +461,11 @@ public void testPausedStart() throws Exception {
 		container.resume();
 
 		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
+		Set<String> clientIds = container.metrics().keySet();
+		assertThat(clientIds).hasSize(2);
+		Iterator<String> iterator = clientIds.iterator();
+		assertThat(iterator.next()).startsWith("myClientId-");
+		assertThat(iterator.next()).startsWith("myClientId-");
 		container.stop();
 		this.logger.info("Stop paused start");
 	}
@@ -636,9 +647,11 @@ private void testAckOnErrorWithManualImmediateGuts(String topic, boolean ackOnEr
 			}
 
 		});
+		containerProps.setClientId("myClientId");
 		ConcurrentMessageListenerContainer<Integer, String> container = new ConcurrentMessageListenerContainer<>(cf,
 				containerProps);
 		container.setConcurrency(1);
+		container.setAlwaysClientIdSuffix(false);
 		container.setBeanName("testAckOnErrorWithManualImmediate");
 		container.start();
 		ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
@@ -651,6 +664,9 @@ private void testAckOnErrorWithManualImmediateGuts(String topic, boolean ackOnEr
 		template.sendDefault(0, 1, "bar");
 		template.flush();
 		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
+		Set<String> clientIds = container.metrics().keySet();
+		assertThat(clientIds).hasSize(1);
+		assertThat(clientIds.iterator().next()).isEqualTo("myClientId");
 		container.stop();
 
 		Consumer<Integer, String> consumer = cf.createConsumer();