[ISSUE-175] Make consumerId unique per thread (#176)

* [ISSUE-175] Make consumerId unique per thread

* Resolve failing tests, fix poor naming

* Update API test

* Update release version, Update Changelog

* fix confusing labels

Author: Crim

CHANGELOG.md

@@ -2,6 +2,10 @@
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
+## 2.3.1 (UNRELEASED)
+#### Bug Fixes
+- [ISSUE-175](https://github.com/SourceLabOrg/kafka-webview/issues/175) Update multi-threaded consumers with unique consumerId [PR](https://github.com/SourceLabOrg/kafka-webview/pull/176).
+
 ## 2.3.0 (06/19/2019)
 #### New Features
 - [ISSUE-166](https://github.com/SourceLabOrg/kafka-webview/issues/166) Add groupSearchFilter property to specify the filter used to list LDAP group membership. Thanks for the contribution @[BlueIcarus](https://github.com/BlueIcarus)!

dev-cluster/pom.xml

@@ -5,12 +5,12 @@
     <parent>
         <artifactId>kafka-webview</artifactId>
         <groupId>org.sourcelab</groupId>
-        <version>2.3.0</version>
+        <version>2.3.1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
     <artifactId>dev-cluster</artifactId>
-    <version>2.3.0</version>
+    <version>2.3.1</version>
 
     <!-- Require Maven 3.3.9 -->
     <prerequisites>

kafka-webview-plugin/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <groupId>org.sourcelab</groupId>
         <artifactId>kafka-webview</artifactId>
-        <version>2.3.0</version>
+        <version>2.3.1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>kafka-webview-plugin</artifactId>

kafka-webview-ui/pom.xml

@@ -5,11 +5,11 @@
     <parent>
         <artifactId>kafka-webview</artifactId>
         <groupId>org.sourcelab</groupId>
-        <version>2.3.0</version>
+        <version>2.3.1</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
     <artifactId>kafka-webview-ui</artifactId>
-    <version>2.3.0</version>
+    <version>2.3.1</version>
 
     <!-- Module Description and Ownership -->
     <name>Kafka WebView UI</name>

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/controller/cluster/ClusterController.java

@@ -168,10 +168,10 @@ public String readTopic(
     /**
      * GET Displays info about a specific consumers group in a cluster.
      */
-    @RequestMapping(path = "/{clusterId}/consumer/{consumerId:.+}", method = RequestMethod.GET)
+    @RequestMapping(path = "/{clusterId}/consumer/{groupId:.+}", method = RequestMethod.GET)
     public String readConsumer(
         @PathVariable final Long clusterId,
-        @PathVariable final String consumerId,
+        @PathVariable final String groupId,
         final Model model,
         final RedirectAttributes redirectAttributes) {
 
@@ -182,12 +182,12 @@ public String readConsumer(
             return "redirect:/";
         }
         model.addAttribute("cluster", cluster);
-        model.addAttribute("consumerId", consumerId);
+        model.addAttribute("groupId", groupId);
 
         // Setup breadcrumbs
         setupBreadCrumbs(model)
             .addCrumb(cluster.getName(), "/cluster/" + clusterId)
-            .addCrumb("Consumer " + consumerId, null);
+            .addCrumb("Consumer Group " + groupId, null);
 
 
         // Display template

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaClientConfigUtil.java

@@ -86,17 +86,22 @@ private Map<String, Object> applyCommonSettings(
         final String consumerId,
         final Map<String, Object> config
     ) {
-        // Generate consumerId with our configured static prefix.
-        final String prefixedConsumerId = consumerIdPrefix.concat("-").concat(consumerId);
+        // Generate groupId with our configured static prefix.
+        final String prefixedGroupId = consumerIdPrefix.concat("-").concat(consumerId);
+
+        // Generate consumerId, which should be unique per user and thread.
+        final String prefixedConsumerId = prefixedGroupId.concat("-") + Thread.currentThread().getId();
 
         // Set common config items
         config.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, clusterConfig.getConnectString());
         config.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, requestTimeoutMs);
 
-        // ClientId and ConsumerGroupId are intended to be unique for each user session.
+        // groupId is intended to be unique for each user session.
+        // clientId is intended to be unique per user session and thread.
         // See Issue-57 https://github.com/SourceLabOrg/kafka-webview/issues/57#issuecomment-363508531
+        // See Issue-175 https://github.com/SourceLabOrg/kafka-webview/issues/175
         config.put(ConsumerConfig.CLIENT_ID_CONFIG, prefixedConsumerId);
-        config.put(ConsumerConfig.GROUP_ID_CONFIG, prefixedConsumerId);
+        config.put(ConsumerConfig.GROUP_ID_CONFIG, prefixedGroupId);
 
         // Optionally configure SSL
         applySslSettings(clusterConfig, config);

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaOperations.java

@@ -440,7 +440,7 @@ public List<ConsumerGroupDetails> getConsumerGroupDetails(final List<String> con
                 });
 
             // Sort list by consumer group id
-            consumerGroupDetails.sort(Comparator.comparing(ConsumerGroupDetails::getConsumerId));
+            consumerGroupDetails.sort(Comparator.comparing(ConsumerGroupDetails::getGroupId));
 
             // Return immutable list.
             return Collections.unmodifiableList(consumerGroupDetails);

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/ParallelWebKafkaConsumer.java

@@ -88,7 +88,7 @@ public ParallelWebKafkaConsumer(
 
     @Override
     public KafkaResults consumePerPartition() {
-        try (final KafkaConsumer kafkaConsumer = createNewConsumer()) {
+        try (final KafkaConsumer kafkaConsumer = createNewConsumerAndSubscribeAllPartitions()) {
             final List<TopicPartition> allTopicPartitions = getAllPartitions(kafkaConsumer);
 
             // To preserve order
@@ -131,7 +131,7 @@ public KafkaResults consumePerPartition() {
 
     @Override
     public ConsumerState seek(final Map<Integer, Long> partitionOffsetMap) {
-        try (final KafkaConsumer kafkaConsumer = createNewConsumer()) {
+        try (final KafkaConsumer kafkaConsumer = createNewConsumerAndSubscribeAllPartitions()) {
             for (final Map.Entry<Integer, Long> entry : partitionOffsetMap.entrySet()) {
                 kafkaConsumer.seek(
                     new TopicPartition(clientConfig.getTopicConfig().getTopicName(), entry.getKey()),
@@ -145,7 +145,7 @@ public ConsumerState seek(final Map<Integer, Long> partitionOffsetMap) {
 
     @Override
     public ConsumerState seek(final long timestamp) {
-        try (final KafkaConsumer kafkaConsumer = createNewConsumer()) {
+        try (final KafkaConsumer kafkaConsumer = createNewConsumerAndSubscribeAllPartitions()) {
             // Find offsets for timestamp
             final Map<TopicPartition, Long> timestampMap = new HashMap<>();
             for (final TopicPartition topicPartition : getAllPartitions(kafkaConsumer)) {
@@ -173,7 +173,7 @@ public void close() {
 
     @Override
     public void previous() {
-        try (final KafkaConsumer kafkaConsumer = createNewConsumer()) {
+        try (final KafkaConsumer kafkaConsumer = createNewConsumerAndSubscribeAllPartitions()) {
             // Get all available partitions
             final List<TopicPartition> topicPartitions = getAllPartitions(kafkaConsumer);
 
@@ -203,7 +203,7 @@ public void previous() {
 
     @Override
     public void next() {
-        try (final KafkaConsumer kafkaConsumer = createNewConsumer()) {
+        try (final KafkaConsumer kafkaConsumer = createNewConsumerAndSubscribeAllPartitions()) {
             // Get all available partitions
             final List<TopicPartition> topicPartitions = getAllPartitions(kafkaConsumer);
 
@@ -231,7 +231,7 @@ public void next() {
 
     @Override
     public ConsumerState toHead() {
-        try (final KafkaConsumer kafkaConsumer = createNewConsumer()) {
+        try (final KafkaConsumer kafkaConsumer = createNewConsumerAndSubscribeAllPartitions()) {
             // Get all available partitions
             final List<TopicPartition> topicPartitions = getAllPartitions(kafkaConsumer);
 
@@ -253,7 +253,7 @@ public ConsumerState toHead() {
 
     @Override
     public ConsumerState toTail() {
-        try (final KafkaConsumer kafkaConsumer = createNewConsumer()) {
+        try (final KafkaConsumer kafkaConsumer = createNewConsumerAndSubscribeAllPartitions()) {
             // Get all available partitions
             final List<TopicPartition> topicPartitions = getAllPartitions(kafkaConsumer);
 
@@ -274,7 +274,20 @@ public ConsumerState toTail() {
         }
     }
 
+    /**
+     * Creates a new consumer, but does NOT subscribe to any partitions.
+     * Will be required to subscribe to the partitions manually you require.
+     * @return KafkaConsumer
+     */
     private KafkaConsumer createNewConsumer() {
+        return kafkaConsumerFactory.createConsumer(clientConfig);
+    }
+
+    /**
+     * Creates a new consumer AND subscribes to all partitions.
+     * @return KafkaConsumer
+     */
+    private KafkaConsumer createNewConsumerAndSubscribeAllPartitions() {
         return kafkaConsumerFactory.createConsumerAndSubscribe(clientConfig);
     }
 

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/manager/kafka/dto/ConsumerGroupDetails.java

@@ -34,7 +34,7 @@
  * Details about a specific consumer group.
  */
 public class ConsumerGroupDetails {
-    private final String consumerId;
+    private final String groupId;
     private final boolean isSimple;
     private final String partitionAssignor;
     private final String state;
@@ -43,30 +43,30 @@ public class ConsumerGroupDetails {
 
     /**
      * Constructor.
-     * @param consumerId consumer group id.
+     * @param groupId consumer group id.
      * @param isSimple if its a simple consumer group
      * @param partitionAssignor How partitions are assigned
      * @param state state of consumer
      * @param members members in the group.
      * @param coordinator node that is acting as the coordinator for this group.
      */
     public ConsumerGroupDetails(
-        final String consumerId,
+        final String groupId,
         final boolean isSimple,
         final String partitionAssignor,
         final String state,
         final List<Member> members,
         final NodeDetails coordinator) {
-        this.consumerId = consumerId;
+        this.groupId = groupId;
         this.isSimple = isSimple;
         this.partitionAssignor = partitionAssignor;
         this.members = Collections.unmodifiableList(new ArrayList<>(members));
         this.state = state;
         this.coordinator = coordinator;
     }
 
-    public String getConsumerId() {
-        return consumerId;
+    public String getGroupId() {
+        return groupId;
     }
 
     public boolean isSimple() {
@@ -142,7 +142,7 @@ public String toString() {
     @Override
     public String toString() {
         return "ConsumerGroupDetails{"
-            + "consumerId='" + consumerId + '\''
+            + "groupId='" + groupId + '\''
             + ", isSimple=" + isSimple
             + ", partitionAssignor='" + partitionAssignor + '\''
             + ", state='" + state + '\''

kafka-webview-ui/src/main/resources/templates/cluster/read.html

@@ -154,7 +154,7 @@
 
                     jQuery.each(ClusterInfo.cachedConsumers, function(index, result) {
                         // Filter
-                        if (!SearchTools.doesMatchText(searchStr, result.consumerId)) {
+                        if (!SearchTools.doesMatchText(searchStr, result.groupId)) {
                             // Skip
                             return;
                         }
@@ -175,8 +175,8 @@
 
                         // Generate html from template
                         var properties = {
-                            id: result.consumerId,
-                            encodedId: encodeURIComponent(result.consumerId),
+                            id: result.groupId,
+                            encodedId: encodeURIComponent(result.groupId),
                             topic: uniqueTopics,
                             state: result.state,
                             numberOfMembers: result.members.length,
@@ -214,11 +214,11 @@
                         ClusterInfo.handleAllConsumers(results);
                     });
                 },
-                removeConsumer: function(consumerId) {
+                removeConsumer: function(groupId) {
                     if (!confirm('Are you sure you want to remove this consumer group?')) {
                         return;
                     }
-                    ApiClient.removeConsumer(ClusterInfo.clusterId, consumerId, function(result) {
+                    ApiClient.removeConsumer(ClusterInfo.clusterId, groupId, function(result) {
                         UITools.showSuccess('Successfully remove consumer group.');
 
                         // Display success