[server] Add pool type information to Kafka consumer client ID (#1541)

A simple fix for Kafka consumer ID generation. Added the pool type so that  
different pool's consumer won't share same client ID. This is helpful when  
debugging consumption with JMX.

clients/da-vinci-client/src/main/java/com/linkedin/davinci/kafka/consumer/KafkaConsumerService.java

@@ -137,7 +137,7 @@ protected KafkaConsumerService(
       /**
        * We need to assign a unique client id across all the storage nodes, otherwise, they will fail into the same throttling bucket.
        */
-      consumerProperties.setProperty(KAFKA_CLIENT_ID_CONFIG, getUniqueClientId(kafkaUrl, i));
+      consumerProperties.setProperty(KAFKA_CLIENT_ID_CONFIG, getUniqueClientId(kafkaUrl, i, poolType));
       SharedKafkaConsumer pubSubConsumer = new SharedKafkaConsumer(
           pubSubConsumerAdapterFactory.create(
               new VeniceProperties(consumerProperties),
@@ -190,8 +190,8 @@ void handleUnsubscription(
       PubSubTopicPartition topicPartition) {
   }
 
-  private String getUniqueClientId(String kafkaUrl, int suffix) {
-    return Utils.getHostName() + "_" + kafkaUrl + "_" + suffix;
+  String getUniqueClientId(String kafkaUrl, int suffix, ConsumerPoolType poolType) {
+    return Utils.getHostName() + "_" + kafkaUrl + "_" + suffix + poolType.getStatSuffix();
   }
 
   @Override

clients/da-vinci-client/src/test/java/com/linkedin/davinci/kafka/consumer/KafkaConsumerServiceTest.java

@@ -94,7 +94,7 @@ public void testTopicWiseGetConsumer() throws Exception {
         ConsumerPoolType.REGULAR_POOL,
         factory,
         properties,
-        1000l,
+        1000L,
         2,
         mock(IngestionThrottler.class),
         mock(KafkaClusterBasedRecordThrottler.class),
@@ -421,6 +421,18 @@ public void testPartitionWiseGetConsumer() {
     Assert.assertEquals(consumerForT1P3, consumerForT2P1);
   }
 
+  @Test
+  public void testGenerateConsumerId() {
+    String hostName = Utils.getHostName();
+    String kafkaUrl = "abc:1234";
+    int suffix = 3;
+    ConsumerPoolType poolType = ConsumerPoolType.SEP_RT_LEADER_POOL;
+    KafkaConsumerService consumerService = mock(KafkaConsumerService.class);
+    doCallRealMethod().when(consumerService).getUniqueClientId(anyString(), anyInt(), any());
+    String expectedResult = hostName + "_" + kafkaUrl + "_" + suffix + poolType.getStatSuffix();
+    Assert.assertEquals(consumerService.getUniqueClientId(kafkaUrl, suffix, poolType), expectedResult);
+  }
+
   @Test
   public void testStoreAwarePartitionWiseGetConsumer() {
     String storeName1 = Utils.getUniqueString("test_consumer_service1");