[ISSUE-57] - Set unique consumerGroupId for each user session (#58)

* [ISSUE-57] - Set unique consumerGroupId for each user session

* Update changelog and default configs

* Fix pom

* add missing test config

Author: Crim

CHANGELOG.md

@@ -2,9 +2,10 @@
 The format is based on [Keep a Changelog](http://keepachangelog.com/)
 and this project adheres to [Semantic Versioning](http://semver.org/).
 
-## 1.0.3 (01/26/2018)
+## 1.0.3 (TBD)
 - Fix 500 error unable to find templates when running under windows.
 - Add start.bat script for running under windows.
+- [Issue#57](https://github.com/SourceLabOrg/kafka-webview/issues/57) Configure consumerId and consumerGroup using a configurable prefix.
 
 ## 1.0.2 (01/26/2018)
 - Increase file upload limit from 15MB to 64MB.

kafka-webview-ui/src/assembly/distribution/config.yml

@@ -6,4 +6,9 @@ security:
 
 ## Various App Configs
 app:
-  key: "SuperSecretKey"
+  ## Should be unique to your installation.
+  ## This key will be used for symmetric encryption of JKS/TrustStore secrets if you configure any SSL enabled Kafka clusters.
+  key: "SuperSecretKey"
+
+  ## Defines a prefix prepended to the Id of all consumers.
+  consumerIdPrefix: "KafkaWebViewConsumer"

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/configuration/AppProperties.java

@@ -45,6 +45,9 @@ public class AppProperties {
     @Value("${app.maxConcurrentWebSocketConsumers}")
     private Integer maxConcurrentWebSocketConsumers = 100;
 
+    @Value("${app.consumerIdPrefix}")
+    private String consumerIdPrefix;
+
     public String getName() {
         return name;
     }
@@ -61,13 +64,18 @@ public Integer getMaxConcurrentWebSocketConsumers() {
         return maxConcurrentWebSocketConsumers;
     }
 
+    public String getConsumerIdPrefix() {
+        return consumerIdPrefix;
+    }
+
     @Override
     public String toString() {
         return "AppProperties{"
             + "name='" + name + '\''
             + ", uploadPath='" + uploadPath + '\''
             + ", appKey='" + appKey + '\''
             + ", maxConcurrentWebSocketConsumers=" + maxConcurrentWebSocketConsumers
+            + ", consumerIdPrefix='" + consumerIdPrefix + '\''
             + '}';
     }
 }

kafka-webview-ui/src/main/java/org/sourcelab/kafka/webview/ui/configuration/PluginConfig.java

@@ -123,6 +123,9 @@ private KafkaAdminFactory getKafkaAdminFactory(final AppProperties appProperties
      * For creating instances of KafkaConsumers.
      */
     private KafkaConsumerFactory getKafkaConsumerFactory(final AppProperties appProperties) {
-        return new KafkaConsumerFactory(appProperties.getUploadPath() + "/keyStores");
+        return new KafkaConsumerFactory(
+            appProperties.getUploadPath() + "/keyStores",
+            appProperties.getConsumerIdPrefix()
+        );
     }
 }

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

@@ -46,14 +46,28 @@
  */
 public class KafkaConsumerFactory {
 
+    /**
+     * Path on filesystem where keystores are persisted.
+     */
     private final String keyStoreRootPath;
 
+    /**
+     * Static prefix to pre-pend to all consumerIds.
+     */
+    private final String consumerIdPrefix;
+
     /**
      * Constructor.
      * @param keyStoreRootPath Parent path to where JKS key/trust stores are saved on disk.
+     * @param consumerIdPrefix Static prefix to pre-pend to all consumerIds.
      */
-    public KafkaConsumerFactory(final String keyStoreRootPath) {
+    public KafkaConsumerFactory(final String keyStoreRootPath, final String consumerIdPrefix) {
+        if (consumerIdPrefix == null) {
+            throw new IllegalArgumentException("ConsumerIdPrefix must be configured!");
+        }
+
         this.keyStoreRootPath = keyStoreRootPath;
+        this.consumerIdPrefix = consumerIdPrefix;
     }
 
     /**
@@ -94,11 +108,18 @@ public KafkaConsumer createConsumerAndSubscribe(final ClientConfig clientConfig)
      * Build an appropriate configuration based on the passed in ClientConfig.
      */
     private Map<String, Object> buildConsumerConfig(final ClientConfig clientConfig) {
+        // Generate consumerId with our configured static prefix.
+        final String prefixedConsumerId = consumerIdPrefix.concat("-").concat(clientConfig.getConsumerId());
+
         // Build config
         final Map<String, Object> configMap = new HashMap<>();
-        configMap.put(ConsumerConfig.CLIENT_ID_CONFIG, clientConfig.getConsumerId());
         configMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, clientConfig.getTopicConfig().getClusterConfig().getConnectString());
 
+        // ClientId and ConsumerGroupId are intended to be unique for each user session.
+        // See Issue-57 https://github.com/SourceLabOrg/kafka-webview/issues/57#issuecomment-363508531
+        configMap.put(ConsumerConfig.CLIENT_ID_CONFIG, prefixedConsumerId);
+        configMap.put(ConsumerConfig.GROUP_ID_CONFIG, prefixedConsumerId);
+
         // Set deserializer classes.
         configMap.put(
             ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,

kafka-webview-ui/src/main/resources/config/base.yml

@@ -46,3 +46,4 @@ app:
   uploadPath: "./data/uploads"
   key: "SuperSecretKey"
   maxConcurrentWebSocketConsumers: 64
+  consumerIdPrefix: "KafkaWebViewConsumer"

kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/manager/kafka/KafkaConsumerFactoryTest.java

@@ -79,7 +79,7 @@ public void testBasicConsumerMultiplePartitions() {
             .produceRecords(maxRecordsPerPoll, topicName, 1);
 
         // Create factory
-        final KafkaConsumerFactory kafkaConsumerFactory = new KafkaConsumerFactory("not/used");
+        final KafkaConsumerFactory kafkaConsumerFactory = new KafkaConsumerFactory("not/used", "TestPrefix");
 
         // Create cluster Config
         final ClusterConfig clusterConfig = ClusterConfig.newBuilder()
@@ -140,7 +140,7 @@ public void testBasicConsumerExcludePartitions() {
             .produceRecords(maxRecordsPerPoll, topicName, 1);
 
         // Create factory
-        final KafkaConsumerFactory kafkaConsumerFactory = new KafkaConsumerFactory("not/used");
+        final KafkaConsumerFactory kafkaConsumerFactory = new KafkaConsumerFactory("not/used", "TestPrefix");
 
         // Create cluster Config
         final ClusterConfig clusterConfig = ClusterConfig.newBuilder()
@@ -205,7 +205,7 @@ public void testBasicConsumerWithRecordFilter() throws InterruptedException {
             .produceRecords(maxRecordsPerPoll, topicName, 1);
 
         // Create factory
-        final KafkaConsumerFactory kafkaConsumerFactory = new KafkaConsumerFactory("not/used");
+        final KafkaConsumerFactory kafkaConsumerFactory = new KafkaConsumerFactory("not/used", "TestPrefix");
 
         // Create cluster Config
         final ClusterConfig clusterConfig = ClusterConfig.newBuilder()
@@ -267,7 +267,7 @@ public void testDeserializerOptions() throws InterruptedException {
             .createTopic(topicName, 1);
 
         // Create factory
-        final KafkaConsumerFactory kafkaConsumerFactory = new KafkaConsumerFactory("not/used");
+        final KafkaConsumerFactory kafkaConsumerFactory = new KafkaConsumerFactory("not/used", "TestPrefix");
 
         // Create cluster Config
         final ClusterConfig clusterConfig = ClusterConfig.newBuilder()

kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/manager/kafka/WebKafkaConsumerFactoryTest.java

@@ -251,7 +251,7 @@ private WebKafkaConsumerFactory createDefaultFactory() {
         final PluginFactory<Deserializer> deserializerPluginFactory = new PluginFactory<>("not/used", Deserializer.class);
         final PluginFactory<RecordFilter> filterPluginFactoryPluginFactory = new PluginFactory<>("not/used", RecordFilter.class);
         final SecretManager secretManager = new SecretManager("Passphrase");
-        final KafkaConsumerFactory kafkaConsumerFactory = new KafkaConsumerFactory("not/used");
+        final KafkaConsumerFactory kafkaConsumerFactory = new KafkaConsumerFactory("not/used", "MyPrefix");
 
         return new WebKafkaConsumerFactory(
             deserializerPluginFactory,

kafka-webview-ui/src/test/java/org/sourcelab/kafka/webview/ui/manager/kafka/WebKafkaConsumerTest.java

@@ -47,7 +47,7 @@ public class WebKafkaConsumerTest {
 
     private final static Logger logger = LoggerFactory.getLogger(WebKafkaConsumerTest.class);
 
-    private final KafkaConsumerFactory kafkaConsumerFactory = new KafkaConsumerFactory("./uploads");
+    private final KafkaConsumerFactory kafkaConsumerFactory = new KafkaConsumerFactory("./uploads", "TestPrefix");
 
     //@Test
     public void doTest() {

kafka-webview-ui/src/test/resources/application.yml

@@ -46,3 +46,4 @@ app:
   uploadPath: "./data/uploads"
   key: "SuperSecretKey"
   maxConcurrentWebSocketConsumers: 64
+  consumerIdPrefix: "KafkaWebViewConsumer"