mBean : AlreadyExistingInstance in kafka Consumer

Happens in 2 places :
1. **Server :** We have 2 consumers ( each for different purpose ) created one after the another with the same client ID - [https://github.com/apache/pinot/blob/master/pinot-core/src/main/java/org/apache/pi[…]ot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java](https://github.com/apache/pinot/blob/master/pinot-core/src/main/java/org/apache/pinot/core/data/manager/realtime/LLRealtimeSegmentDataManager.java#L1454) . _Potential solution :_ To Add caller class name - PartitionGroupConsumer , PartitionMetaDataConsumer to it . , also including the
segmentName provides us more granular info and less chance of AlreadyExisting Exception.
```
javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=transcript2_REALTIME-transcript-topic-0
at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:436) ~[?:?]
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1855) ~[?:?]
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:955) ~[?:?]
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:890) ~[?:?]
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:320) ~[?:?]
at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522) ~[?:?]
at org.apache.pinot.shaded.org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:64) [pinot-protobuf-0.13.0-SNAPSHOT-shaded.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.shaded.org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:814) [pinot-protobuf-0.13.0-SNAPSHOT-shaded.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.shaded.org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:665) [pinot-protobuf-0.13.0-SNAPSHOT-shaded.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.shaded.org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:646) [pinot-protobuf-0.13.0-SNAPSHOT-shaded.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.shaded.org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:626) [pinot-protobuf-0.13.0-SNAPSHOT-shaded.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.plugin.stream.kafka20.KafkaPartitionLevelConnectionHandler.createConsumer(KafkaPartitionLevelConnectionHandler.java:84) [pinot-kafka-2.0-0.13.0-SNAPSHOT-shaded.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.plugin.stream.kafka20.KafkaPartitionLevelConnectionHandler.(KafkaPartitionLevelConnectionHandler.java:70) [pinot-kafka-2.0-0.13.0-SNAPSHOT-shaded.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.plugin.stream.kafka20.KafkaStreamMetadataProvider.(KafkaStreamMetadataProvider.java:54) [pinot-kafka-2.0-0.13.0-SNAPSHOT-shaded.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory.createPartitionMetadataProvider(KafkaConsumerFactory.java:43) [pinot-kafka-2.0-0.13.0-SNAPSHOT-shaded.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.core.data.manager.realtime.LLRealtimeSegmentDataManager.createPartitionMetadataProvider(LLRealtimeSegmentDataManager.java:1612) [pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.core.data.manager.realtime.LLRealtimeSegmentDataManager.(LLRealtimeSegmentDataManager.java:1460) [pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.core.data.manager.realtime.RealtimeTableDataManager.addSegment(RealtimeTableDataManager.java:454) [pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.server.starter.helix.HelixInstanceDataManager.addRealtimeSegment(HelixInstanceDataManager.java:249) [pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.server.starter.helix.SegmentOnlineOfflineStateModelFactory$SegmentOnlineOfflineStateModel.onBecomeOnlineFromOffline(SegmentOnlineOfflineStateModelFactory.java:168) [pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.server.starter.helix.SegmentOnlineOfflineStateModelFactory$SegmentOnlineOfflineStateModel.onBecomeConsumingFromOffline(SegmentOnlineOfflineStateModelFactory.java:83) [pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:?]
at jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:?]
at jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:?]
at java.lang.reflect.Method.invoke(Method.java:566) ~[?:?]
at org.apache.helix.messaging.handling.HelixStateTransitionHandler.invoke(HelixStateTransitionHandler.java:350) [pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.helix.messaging.handling.HelixStateTransitionHandler.handleMessage(HelixStateTransitionHandler.java:278) [pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.helix.messaging.handling.HelixTask.call(HelixTask.java:97) [pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.helix.messaging.handling.HelixTask.call(HelixTask.java:49) [pinot-all-0.13.0-SNAPSHOT-jar-with-dependencies.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
at java.lang.Thread.run(Thread.java:829) [?:?]`
```

2. **Controller :** The period Job to check the segmentStatus , eventually makes KafkaConsumer , it has a retry mechanism to create a paritionGroupMetaDataFetcher , which is causing issue in creating multiple kafka consumer of the same name. _Potential Solution :_ Include a incremental ID to the clientID in the retry code .

```
javax.management.InstanceAlreadyExistsException: kafka.consumer:type=app-info,id=PartitionGroupMetadataFetcher-transcript_REALTIME-transcript-topic
--
at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:436) ~[?:?]
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1855) ~[?:?]
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:955) ~[?:?]
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:890) ~[?:?]
at com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:320) ~[?:?]
at com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522) ~[?:?]
at org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:64) ~[kafka-clients-2.8.1.jar:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:814) ~[kafka-clients-2.8.1.jar:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:665) ~[kafka-clients-2.8.1.jar:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:646) ~[kafka-clients-2.8.1.jar:?]
at org.apache.kafka.clients.consumer.KafkaConsumer.(KafkaConsumer.java:626) ~[kafka-clients-2.8.1.jar:?]
at org.apache.pinot.plugin.stream.kafka20.KafkaPartitionLevelConnectionHandler.createConsumer(KafkaPartitionLevelConnectionHandler.java:84) ~[pinot-kafka-2.0-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.plugin.stream.kafka20.KafkaPartitionLevelConnectionHandler.(KafkaPartitionLevelConnectionHandler.java:70) ~[pinot-kafka-2.0-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.plugin.stream.kafka20.KafkaStreamMetadataProvider.(KafkaStreamMetadataProvider.java:54) ~[pinot-kafka-2.0-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.plugin.stream.kafka20.KafkaConsumerFactory.createPartitionMetadataProvider(KafkaConsumerFactory.java:43) ~[pinot-kafka-2.0-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-02fe2f1f864627b4a0e80c1291422b75cd173879]
at org.apache.pinot.spi.stream.StreamMetadataProvider.computePartitionGroupMetadata(StreamMetadataProvider.java:83) ~[pinot-spi-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.spi.stream.PartitionGroupMetadataFetcher.call(PartitionGroupMetadataFetcher.java:70) ~[pinot-spi-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.spi.stream.PartitionGroupMetadataFetcher.call(PartitionGroupMetadataFetcher.java:31) ~[pinot-spi-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.spi.utils.retry.BaseRetryPolicy.attempt(BaseRetryPolicy.java:50) ~[pinot-spi-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.controller.helix.core.PinotTableIdealStateBuilder.getPartitionGroupMetadataList(PinotTableIdealStateBuilder.java:160) ~[pinot-controller-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.controller.helix.core.realtime.MissingConsumingSegmentFinder.(MissingConsumingSegmentFinder.java:79) ~[pinot-controller-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.controller.helix.SegmentStatusChecker.updateSegmentMetrics(SegmentStatusChecker.java:316) ~[pinot-controller-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.controller.helix.SegmentStatusChecker.processTable(SegmentStatusChecker.java:122) ~[pinot-controller-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.controller.helix.SegmentStatusChecker.processTable(SegmentStatusChecker.java:61) ~[pinot-controller-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask.processTables(ControllerPeriodicTask.java:116) ~[pinot-controller-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.controller.helix.core.periodictask.ControllerPeriodicTask.runTask(ControllerPeriodicTask.java:85) ~[pinot-controller-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.core.periodictask.BasePeriodicTask.run(BasePeriodicTask.java:150) ~[pinot-core-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.core.periodictask.BasePeriodicTask.run(BasePeriodicTask.java:135) ~[pinot-core-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at org.apache.pinot.core.periodictask.PeriodicTaskScheduler.lambda$start$0(PeriodicTaskScheduler.java:87) ~[pinot-core-0.13.0-SNAPSHOT.jar:0.13.0-SNAPSHOT-1b0199e7ec7eb5c6afe0ae735ce4946241154b8c]
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) [?:?]
at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) [?:?]
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) [?:?]
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) [?:?]
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) [?:?]
at java.lang.Thread.run(Thread.java:829) [?:?]
```

This is from latest master branch .

Related PR : https://github.com/apache/pinot/pull/10324

cc : @ankitsultana @Aravind-Suresh

Start with LLRealtimeSegmentDataManager.java and KafkaPartitionLevelConnectionHandler.java to trace the server-side consumer creation, then follow PinotTableIdealStateBuilder.java and PartitionGroupMetadataFetcher for controller retries. Verify how client IDs are formed in both paths and ensure repeated consumer creation no longer produces duplicate JMX names; confirm both reported stack-trace scenarios are covered.

Written by the indexing model from the issue text.