1.1-5

Author: anders-wartoft

README.md

@@ -18,6 +18,11 @@ java -jar target/LogGenerator{version}.jar -i kafka -ci test3 -t OUTPUT -b 192.1
 When running the last command, press Ctrl-C to see the gaps in the received data. Since we started the counter on 100, there should at least be one gap: 1-99.
 
 ### Latest Release Notes
+
+### 1.1-5
+Added a property --print-keys (-pk) to the KafkaInputItem to print the Kafka key for lines read.
+Added a property --start-number (-sn) to the CounterInputItem to alter the start number 
+
 #### 1.1-4
 Changed kafka-clients dependency version from 3.7.1 to 3.9.1 due to CVE-2025-27817
 
@@ -162,7 +167,7 @@ Parameters and example, see below in the Q&A section.
 #### Fetch from Kafka topics
 Connect to a Kafka server and read from a topic
 
-Parameters: `-i kafka -ci {client id} -t {topic name} -b {boostrap server}`
+Parameters: `-i kafka -ci {client id} -t {topic name} -b {boostrap server} --print-keys {true/false}`
 
 Example: `-i kafka -ci test -t testtopic -b localhost:9092`
 
@@ -174,7 +179,7 @@ The main use case is to get one field from an index and send to a receiver, for
 Parameters: `-i elastic --hostname {hostname or ip} -p {port number} -i {index name} -f {field to get} -ak {API key} -c {x.509 certificate in cer format} -q {query string in query dsl format}`
 
 Example:
-``` properties
+```ini
 # Elasticsearch
 input=elastic
 # Elastic instance to connect to
@@ -295,7 +300,7 @@ The main use case is to be able to generate events with _id set to some enumerab
 
 Parameters `-o elastic --hostname {hostname or ip} --port {port number} --index {index} --api-key {API Key} --regex {regex to find an id from input} --id {format for output id} --certificate-path {X.509 certificate for the elastic server in cer format}`
 
-``` properties
+```ini
 # Elasticsearch
 output=elastic
 # Elastic instance to connect to
@@ -728,7 +733,7 @@ A combination of parameters from the command line and property file is possible.
 The property file can have all short- or long names for configuration. Comment lines start with the hash character '#'.
 
 Example:
-```properties
+```ini
 # Start an udp proxy, listening on port 9999, rewriting dates to the current date. 
 # Dates are parsed  according to a specified format and finally sending the result
 # to a syslog server.
@@ -774,15 +779,15 @@ There are three ways to add custom variables:
 
 Example:
 Say you would like to always exchange the text `{company-name}` in a template with `sitia.nu`. Then add the following to a property file (let's call it variables.properties):
-```properties
+```ini
 company-name=sitia.nu
 ```
 and add that file to the property file loaded by `-pf` on the command to LogGenerator:
-```properties
+```ini
 variable-file=variables.properties
 ```
 If you only have a small number of custom variables, you could just load all of them in the property file loaded by `-p`:
-```properties
+```ini
 custom.company-name=sitia.nu
 ```
 Custom variables can contain variables.
@@ -1071,7 +1076,7 @@ Also, you can omit the escape:
 -f gap --regex "<(\d+)>"
 ```
 But, in the property file you have to write the regex as is, without extra escape characters, like
-```properties
+```ini
 filter=gap
 regex=<(\d+)>
 ```

dependency-reduced-pom.xml

@@ -4,7 +4,7 @@
   <groupId>nu.sitia.LogGenerator</groupId>
   <artifactId>LogGenerator</artifactId>
   <name>LogGenerator</name>
-  <version>1.1-4</version>
+  <version>1.1-5</version>
   <url>http://maven.apache.org</url>
   <build>
     <plugins>

pom.xml

@@ -4,7 +4,7 @@
 
   <groupId>nu.sitia.LogGenerator</groupId>
   <artifactId>LogGenerator</artifactId>
-  <version>1.1-4</version>
+  <version>1.1-5</version>
   <packaging>jar</packaging>
 
   <name>LogGenerator</name>

src/main/java/nu/sitia/loggenerator/inputitems/CounterInputItem.java

@@ -46,8 +46,9 @@ public boolean setParameter(String key, String value) {
             System.out.println("CounterInputItem. Return a string with an ever-increasing number\n" +
                     "Parameters:\n" +
                     "--string <string> (-st <string>)\n" +
-                    "  The string to return\n");
-            super.setParameter(key, value);
+                    "  The string to return\n" +
+                    "--start-number <number> (-sn <number>)\n" +
+                    "  The number to start with (default 1)\n");
         }
         if(super.setParameter(key, value)) {
             return true;
@@ -56,6 +57,14 @@ public boolean setParameter(String key, String value) {
             this.string = value;
             return true;
         }
+        if (key != null && (key.equalsIgnoreCase("--start-number") || key.equalsIgnoreCase("-sn"))) {
+            try {
+                this.number = Long.parseLong(value);
+            } catch (NumberFormatException e) {
+                System.out.println("Invalid number format for --start-number");
+            }
+            return true;
+        }
         return false;
     }
 

src/main/java/nu/sitia/loggenerator/inputitems/KafkaInputItem.java

@@ -19,6 +19,7 @@
 
 import nu.sitia.loggenerator.Configuration;
 import org.apache.kafka.clients.consumer.*;
+import org.apache.kafka.common.errors.UnknownTopicIdException;
 
 import java.time.Duration;
 import java.util.ArrayList;
@@ -45,6 +46,10 @@ public class KafkaInputItem extends AbstractInputItem {
     /** The topic name */
     private String topicName;
 
+    /** Whether to print the keys read from Kafka */
+    private boolean printKeys = false;
+
+    /** The default Kafka batch size */
     private final static int DEFAULT_KAFKA_BATCHSIZE = 200;
 
     /**
@@ -65,7 +70,9 @@ public boolean setParameter(String key, String value) {
                     "--topic <topic> (-t <topic>)\n" +
                     "  The topic to read from\n" +
                     "--bootstrap-server <hostname:port> (-b <hostname:port>)\n" +
-                    "  The hostname:port to connect to\n");
+                    "  The hostname:port to connect to\n" +
+                    "--print-keys <true|false> (-pk <true|false>)\n" +
+                    "  Whether to print the keys read from Kafka. Default false\n");
             super.setParameter(key, value);
         }
         if (super.setParameter(key, value)) {
@@ -86,6 +93,11 @@ public boolean setParameter(String key, String value) {
             logger.fine("bootstrapServer " + value);
             return true;
         }
+        if (key != null && (key.equalsIgnoreCase("--print-keys") || key.equalsIgnoreCase("-pk"))) {
+            this.printKeys = Boolean.parseBoolean(value);
+            logger.fine("printKeys " + value);
+            return true;
+        }
         return false;
     }
 
@@ -114,10 +126,12 @@ public void setup() throws RuntimeException {
         properties.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, String.valueOf(batchSize));
         properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
         properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
+        properties.put(ConsumerConfig.METADATA_MAX_AGE_CONFIG, "1000");
 
         consumer = new KafkaConsumer<>(properties);
         logger.info("Connected to kafka " + this.bootstrapServer);
         consumer.subscribe(Collections.singletonList(this.topicName));
+        consumer.poll(Duration.ofSeconds(5)); // fetch metadata and leader info
         logger.info("Subscribed to topic " + this.topicName);
     }
 
@@ -136,13 +150,28 @@ public boolean hasNext() {
      * @return The read input
      */
     public List<String> next() {
-        ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofMillis(1000));
         List<String> result = new ArrayList<>();
-        for (ConsumerRecord<Integer, String> record: records) {
-            result.add(record.value());
+        while (true) {
+            try {
+                ConsumerRecords<Integer, String> records = consumer.poll(Duration.ofMillis(1000));
+                for (ConsumerRecord<Integer, String> record: records) {
+                    if (printKeys) {
+                        result.add(record.key() + ": " + record.value());
+                    } else {
+                        result.add(record.value());
+                    }
+                }
+                logger.log(Level.FINEST, result.toString());
+                return result;
+            } catch (UnknownTopicIdException e) {
+                try {
+                    // wait a bit and retry
+                    Thread.sleep(500);
+                } catch (InterruptedException e1) {
+                    return result;
+                }
+            }
         }
-        logger.log(Level.FINEST, result.toString());
-        return result;
     }
 
     /**