Add topic prefix flag

.github/workflows/integration.yaml

@@ -34,5 +34,20 @@ jobs:
       - name: Produce to Kafka from Avro Schema in Avro Format
         run: docker exec datagen datagen -s /tests/schema.avsc -f avro -n 3 --record-size 100 -d -w 100
 
+      - name: Topic prefix with json
+        run: docker exec datagen datagen -s /tests/schema.sql -f json -n 3 --record-size 100 -d --prefix test
+
+      - name: Topic prefix with avro
+        run: docker exec datagen datagen -s /tests/schema.sql -f avro -n 3 --record-size 100 -d --prefix test
+
+      - name: Clean Kafka topic
+        run: docker exec datagen datagen -s /tests/schema.sql -f json -d --clean
+
+      - name: Clean Kafka topic with prefix
+        run: docker exec datagen datagen -s /tests/schema.sql -f json -d --clean --prefix test
+
+      - name: Clean Kafka topic and schema registry
+        run: docker exec datagen datagen -s /tests/schema.avsc -f avro -d --clean
+
       - name: Docker Compose Down
         run: docker compose down -v

datagen.js

@@ -36,7 +36,8 @@ program
     .option('-dr, --dry-run', 'Dry run (no data will be produced to Kafka)')
     .option('-d, --debug', 'Output extra debugging information')
     .option('-w, --wait <int>', 'Wait time in ms between record production', parseInt)
-    .option('-rs, --record-size <int>', 'Record size in bytes, eg. 1048576 for 1MB', parseInt);
+    .option('-rs, --record-size <int>', 'Record size in bytes, eg. 1048576 for 1MB', parseInt)
+    .option('-p, --prefix <char>', 'Kafka topic and schema registry prefix');
 
 program.parse();
 
@@ -56,6 +57,7 @@ global.recordSize = options.recordSize;
 global.wait = options.wait;
 global.clean = options.clean;
 global.dryRun = options.dryRun;
+global.prefix = options.prefix;
 
 if (debug) {
     console.log(options);

src/dataGenerator.js

@@ -1,8 +1,8 @@
 const alert = require('cli-alerts');
 const crypto = require('crypto');
 const createTopic = require('./kafka/createTopic');
-const producer = require('./kafka/producer');
 const schemaRegistryConfig = require('./kafka/schemaRegistryConfig');
+const {kafkaProducer, connectKafkaProducer, disconnectKafkaProducer} = require('./kafka/producer');
 const {
     getAvroEncodedRecord,
     registerSchema,
@@ -93,6 +93,7 @@ module.exports = async ({
     schema,
     number
 }) => {
+
     let payload;
     if (recordSize) {
         recordSize = recordSize / 2;
@@ -101,7 +102,9 @@ module.exports = async ({
 
     let registry;
     let avroSchemas = {};
-
+    if(dryRun !== true){
+        const producer = await connectKafkaProducer();
+    }
     for await (const iteration of asyncGenerator(number)) {
         global.iterationIndex = iteration;
         megaRecord = await generateMegaRecord(schema);
@@ -151,11 +154,14 @@ module.exports = async ({
                             avroSchemas[topic]['schemaId']
                         );
                     }
-                    await producer(recordKey, record, encodedRecord, topic);
+                    await kafkaProducer(producer, recordKey, record, encodedRecord, topic);
                 }
             }
         }
 
         await sleep(wait);
     }
+    if(dryRun !== true){
+        await disconnectKafkaProducer(producer);
+    }
 };

src/kafka/cleanKafka.js

@@ -1,6 +1,7 @@
 const kafkaConfig = require('./kafkaConfig');
 const axios = require('axios');
 const dotenv = require('dotenv');
+const alert = require('cli-alerts');
 
 
 async function deleteSchemaSubjects(topics) {
@@ -10,6 +11,7 @@ async function deleteSchemaSubjects(topics) {
         process.exit();
     }
     for await (const topic of topics){
+
         let url = `${process.env.SCHEMA_REGISTRY_URL}/subjects/${topic}-value?permanent=false`;
         await axios.delete(
             url,
@@ -36,6 +38,16 @@ module.exports = async (format, topics) => {
         console.log("This is a dry run, so no resources will be deleted")
         return
     }
+    if (prefix) {
+        // Loop through topics and add prefix
+        topics = topics.map(topic => `${prefix}_${topic}`);
+        alert({
+            type: `success`,
+            name: `Using topic with prefix: ${prefix}`,
+            msg: ``
+        });
+    }
+
     const kafka = kafkaConfig();
     const admin = kafka.admin();
     await admin.connect();

src/kafka/createTopic.js

@@ -1,5 +1,6 @@
 const { ConfigResourceTypes } = require('kafkajs');
 const kafkaConfig = require('./kafkaConfig');
+const alert = require('cli-alerts');
 const dotenv = require('dotenv');
 
 module.exports = async (topic = 'datagen_test_topic') => {
@@ -8,6 +9,16 @@ module.exports = async (topic = 'datagen_test_topic') => {
     if (debug) {
         console.log(`Trying to create topic: ${topic}`);
     }
+
+    if (prefix) {
+        topic = `${prefix}_${topic}`;
+        alert({
+            type: `success`,
+            name: `Using topic with prefix: ${topic}`,
+            msg: ``
+        });
+    }
+
     // Check if the topic exists in the Kafka cluster if not create it
     const admin = kafka.admin();
     await admin.connect();

src/kafka/producer.js

@@ -3,14 +3,16 @@ const kafkaConfig = require('./kafkaConfig');
 const alert = require('cli-alerts');
 const dotenv = require('dotenv');
 
-module.exports = async (recordKey = null, record, encodedRecord = null, topic = 'datagen_test_topic') => {
-    // Produce the record to Kafka
-    const kafka = kafkaConfig();
+async function kafkaProducer(producer, recordKey = null, record, encodedRecord = null, topic = 'datagen_test_topic') {
 
-    const producer = kafka.producer({
-        createPartitioner: Partitioners.DefaultPartitioner
-    });
-    await producer.connect();
+    if (prefix) {
+        topic = `${prefix}_${topic}`;
+        alert({
+            type: `success`,
+            name: `Using topic with prefix: ${topic}`,
+            msg: ``
+        });
+    }
 
     let payload;
     if (encodedRecord) {
@@ -36,6 +38,28 @@ module.exports = async (recordKey = null, record, encodedRecord = null, topic =
         name: `Record sent to Kafka topic: ${topic}`,
         msg: `\nkey: ${recordKey}\nvalue:\n${JSON.stringify(record)}`
     });
+};
+
+async function connectKafkaProducer() {
+    const kafka = kafkaConfig();
+    const producer = kafka.producer({
+        createPartitioner: Partitioners.DefaultPartitioner
+    });
 
+    if (debug) {
+        console.log(`Connecting to Kafka producer...`);
+    }
+    await producer.connect();
+    return producer;
+}
+
+async function disconnectKafkaProducer(producer) {
+    if (debug) {
+        console.log(`Disconnecting from Kafka producer...`);
+    }
     await producer.disconnect();
-};
+}
+
+module.exports.kafkaProducer = kafkaProducer;
+module.exports.connectKafkaProducer = connectKafkaProducer;
+module.exports.disconnectKafkaProducer = disconnectKafkaProducer;