final prep before release

Author: Stephen Powis

dev-cluster/pom.xml

@@ -111,6 +111,20 @@
                     <skip>true</skip>
                 </configuration>
             </plugin>
+
+            <!-- Copy dependencies over into packaged jar -->
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-dependency-plugin</artifactId>
+                <version>3.1.1</version>
+                <executions>
+                    <execution>
+                        <id>copy-dependencies</id>
+                        <phase>package</phase>
+                        <goals><goal>copy-dependencies</goal></goals>
+                    </execution>
+                </executions>
+            </plugin>
         </plugins>
     </build>
 

dev-cluster/src/main/java/org/sourcelab/kafka/devcluster/DevCluster.java

@@ -38,8 +38,10 @@
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.Options;
 import org.apache.commons.cli.ParseException;
+import org.apache.kafka.clients.admin.TopicDescription;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartitionInfo;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -141,12 +143,24 @@ public static void main(final String[] args) throws Exception {
                 logger.info("Started broker with Id {} at {}", broker.getBrokerId(), broker.getConnectString());
             });
 
+        if (topicNames != null && topicNames.length > 0) {
+            final KafkaTestUtils testUtils = new KafkaTestUtils(kafkaTestCluster);
+            if (cmd.hasOption("consumer")) {
+                for (final String topicName : topicNames) {
+                    runEndlessConsumer(topicName, testUtils);
+                }
+            }
+
+            if (cmd.hasOption("producer")) {
+                for (final String topicName : topicNames) {
+                    runEndlessProducer(topicName, testUtils);
+                }
+            }
+        }
+
         // Log cluster connect string.
         logger.info("Cluster started at: {}", kafkaTestCluster.getKafkaConnectString());
 
-        //runEndlessConsumer(topicName, utils);
-        //runEndlessProducer(topicName, partitionsCount, utils);
-
         // Wait forever.
         Thread.currentThread().join();
     }
@@ -189,6 +203,24 @@ private static CommandLine parseArguments(final String[] args) throws ParseExcep
             .build()
         );
 
+        // Optionally start an endless consumer
+        options.addOption(Option.builder("consumer")
+            .desc("Start an endless consumer for each of the defined topics.")
+            .required(false)
+            .hasArg(false)
+            .type(Boolean.class)
+            .build()
+        );
+
+        // Optionally start an endless consumer
+        options.addOption(Option.builder("producer")
+            .desc("Start an endless producer for each of the defined topics.")
+            .required(false)
+            .hasArg(false)
+            .type(Boolean.class)
+            .build()
+        );
+
         try {
             final CommandLineParser parser = new DefaultParser();
             return parser.parse(options, args);
@@ -206,20 +238,20 @@ private static CommandLine parseArguments(final String[] args) throws ParseExcep
     /**
      * Fire up a new thread running an endless producer script into the given topic and partitions.
      * @param topicName Name of the topic to produce records into.
-     * @param partitionCount number of partitions that exist on that topic.
      * @param utils KafkaUtils instance.
      */
     private static void runEndlessProducer(
         final String topicName,
-        final int partitionCount,
         final KafkaTestUtils utils
     ) {
         final Thread producerThread = new Thread(() -> {
-            do {
+            // Determine how many partitions there are
+            final TopicDescription topicDescription = utils.describeTopic(topicName);
 
-                // Publish some data into that topic
-                for (int partition = 0; partition < partitionCount; partition++) {
-                    utils.produceRecords(1000, topicName, partition);
+            do {
+                // Publish some data into that topic for each partition.
+                for (final TopicPartitionInfo partitionInfo : topicDescription.partitions()) {
+                    utils.produceRecords(1000, topicName, partitionInfo.partition());
                     try {
                         Thread.sleep(1000L);
                     } catch (InterruptedException e) {
@@ -233,8 +265,10 @@ private static void runEndlessProducer(
                 }
             } while (true);
         });
-        producerThread.start();
 
+        logger.info("Starting endless producer for topic {}", topicName);
+        producerThread.setName("Endless producer for topic " + topicName);
+        producerThread.start();
     }
 
     /**
@@ -270,6 +304,9 @@ private static void runEndlessConsumer(final String topicName, final KafkaTestUt
                 return;
             }
         });
+
+        logger.info("Starting endless consumer for topic {}", topicName);
+        consumerThread.setName("Endless consumer for topic " + topicName);
         consumerThread.start();
     }
 }

dev-cluster/src/main/java/org/sourcelab/kafka/devcluster/LdapServer.java

@@ -42,9 +42,11 @@ public class LdapServer {
      * @param args command line arguments (none).
      */
     public static void main(final String[] args) throws LDAPException, InterruptedException {
+        final String baseDn = "dc=springframework,dc=org";
+
         // Create a base configuration for the server.
         final InMemoryDirectoryServerConfig config = new InMemoryDirectoryServerConfig(
-            "dc=springframework,dc=org"
+            baseDn
         );
 
         // Create new listener
@@ -66,9 +68,10 @@ public static void main(final String[] args) throws LDAPException, InterruptedEx
         server.startListening();
 
         logger.info(
-            "Server running at: {}:{}",
+            "Server running at: ldap://{}:{}/{}",
             server.getConnection("LDAP").getConnectedAddress(),
-            server.getConnection("LDAP").getConnectedPort()
+            server.getConnection("LDAP").getConnectedPort(),
+            baseDn
         );
 
         Thread.currentThread().join();

dev-cluster/startDevCluster.sh

@@ -0,0 +1,28 @@
+#!/bin/bash
+
+set -e
+
+TRUST_STORE_FILE="src/main/resources/kafka.truststore.jks"
+KEY_STORE_FILE="src/main/resources/kafka.keystore.jks"
+JAR_FILE="target/dev-cluster-2.1.0.jar"
+JAAS_FILE="src/main/resources/jaas.conf"
+
+## Change to local directory
+cd "${0%/*}"
+
+## Ensure certificates exist
+if [ ! -f $KEY_STORE_FILE ] || [ ! -f $TRUST_STORE_FILE ]; then
+    echo "Key store files do no exist...generating now...";
+    ## Build certificates
+    ././../generateDummySslCertificates.sh
+fi
+
+## If jar doesn't exit, build it.
+if [ ! -f $JAR_FILE ]; then
+    echo "Building package...";
+    mvn clean package -DskipTests=true -DskipLicenseCheck=true
+fi
+
+## Execute.
+echo "starting..."
+java -Djava.security.auth.login.config=$JAAS_FILE -cp "${JAR_FILE}:target/dependency/*" org.sourcelab.kafka.devcluster.DevCluster "$@"

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

@@ -27,40 +27,41 @@ app:
     ## Require user authentication
     ## Setting to false will disable login requirement.
     enabled: true
-      ## Optional: if you want to use LDAP for user authentication instead of locally defined users.
-      ldap:
-        ## Disabled by default.
-        enabled: false
 
-        ## Example values defined below, adjust as needed.
-        ## How to find user records
-        userDnPattern: "uid={0},ou=people"
+    ## Optional: if you want to use LDAP for user authentication instead of locally defined users.
+    ldap:
+      ## Disabled by default.
+      enabled: false
 
-        ## The attribute in which the password is stored.
-        passwordAttribute: "userPassword"
+      ## URL/Hostname for your LDAP server
+      url: "ldap://localhost:8389/dc=example,dc=org"
 
-        ## Where to find user group membership
-        groupSearchBase: "ou=groups"
-        groupRoleAttribute: "cn"
+      ## Example values defined below, adjust as needed.
+      ## How to find user records
+      userDnPattern: "uid={0},ou=people"
 
-        ## How passwords are validated, must implement PasswordEncoder interface
-        passwordEncoderClass: "org.springframework.security.crypto.password.LdapShaPasswordEncoder"
+      ## The attribute in which the password is stored.
+      passwordAttribute: "userPassword"
 
-        ## Comma separated list of groups. A user which is a member of this group will be granted
-        ## administrator access to Kafka WebView.
-        adminGroups: "ADMINGROUP1,ADMINGROUP2"
+      ## Where to find user group membership
+      groupSearchBase: "ou=groups"
+      groupRoleAttribute: "cn"
 
-        ## Comma separated list of groups. A user which is a member of this group will be granted
-        ## standard user level access to Kafka WebView.
-        userGroups: "USERGROUP1,USERGROUP2"
+      ## How passwords are validated, must implement PasswordEncoder interface
+      passwordEncoderClass: "org.springframework.security.crypto.password.LdapShaPasswordEncoder"
 
-        ## Any user who is not a member of at least one of the above groups will be denied access
-        ## to Kafka WebView.
+      ## Comma separated list of groups. A user which is a member of this group will be granted
+      ## administrator access to Kafka WebView.
+      adminGroups: "ADMINGROUP1,ADMINGROUP2"
 
-        ## URL/Hostname for your LDAP server
-        url: "ldap://localhost:8389/dc=example,dc=org"
+      ## Comma separated list of groups. A user which is a member of this group will be granted
+      ## standard user level access to Kafka WebView.
+      userGroups: "USERGROUP1,USERGROUP2"
 
-        ## If LDAP does not allow anonymous access, define the user/password to connect using.
-        ## If not required, leave both fields empty
-        bindUser: "cn=ManagementUser"
-        bindUserPassword: "password-here"
+      ## Any user who is not a member of at least one of the above groups will be denied access
+      ## to Kafka WebView.
+
+      ## If LDAP does not allow anonymous access, define the user/password to connect using.
+      ## If not required, leave both fields empty
+      bindUser: "cn=ManagementUser"
+      bindUserPassword: "password-here"

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

@@ -309,8 +309,6 @@
                         } else {
                             consumerPositions.push(0);
                         }
-                        //tailPositions.push(offsetData.tail);
-                        //consumerPositions.push(offsetData.offset);
                     });
 
                     // Push data into graph sources

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

@@ -37,7 +37,7 @@
                             </thead>
                             <tbody>
                                 <tr th:if="${clusterList.isEmpty()}" align="center">
-                                    <td colspan="5">
+                                    <td colspan="6">
                                         No clusters found!
                                     </td>
                                 </tr>