GH-1485: Option to suppress single client.id

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 true to suppress.
+	 * @since 2.2.14
+	 */
+	public void setAlwaysClientIdSuffix(boolean alwaysClientIdSuffix) {
+		this.alwaysClientIdSuffix = alwaysClientIdSuffix;
+	}
+
 	/**
 	 * Return the list of {@link KafkaMessageListenerContainer}s created by
 	 * this container.
@@ -181,7 +193,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

@@ -28,6 +28,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.Properties;
@@ -453,6 +454,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);
@@ -467,6 +469,9 @@ public void testManualCommitSyncExisting() throws Exception {
 		template.flush();
 		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
 		assertThat(bitSet.cardinality()).isEqualTo(8);
+		Set<String> clientIds = container.getAssignmentsByClientId().keySet();
+		assertThat(clientIds).hasSize(1);
+		assertThat(clientIds.iterator().next()).isEqualTo("myClientId-0");
 		container.stop();
 		this.logger.info("Stop MANUAL_IMMEDIATE with Existing");
 	}
@@ -483,10 +488,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();
@@ -505,6 +511,11 @@ public void testPausedStart() throws Exception {
 		container.resume();
 
 		assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
+		Set<String> clientIds = container.getAssignmentsByClientId().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");
 	}
@@ -704,9 +715,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());
@@ -719,6 +732,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.getAssignmentsByClientId().keySet();
+		assertThat(clientIds).hasSize(1);
+		assertThat(clientIds.iterator().next()).isEqualTo("myClientId");
 		container.stop();
 
 		Consumer<Integer, String> consumer = cf.createConsumer();

src/reference/asciidoc/kafka.adoc

@@ -2384,6 +2384,11 @@ Suffix
 | Default
 | Description
 
+|alwaysClientId
+Suffix
+|`true`
+|Set to false to suppress adding a suffix to the `client.id` consumer property, when the `concurrency` is only 1.
+
 |assigned
 Partitions
 |(read only)