clean topics and schema subjects datagen creates (#55)

* implement feature to clean topics and schema subjects datagen creates

* better error handling and soft delete the subject

* limit the axios error message for schema registry

* log the names of the deleted topics

datagen.js

@@ -12,7 +12,8 @@ const alert = require('cli-alerts');
 const { parseSqlSchema } = require('./src/schemas/parseSqlSchema');
 const { parseAvroSchema } = require('./src/schemas/parseAvroSchema');
 const { parseJsonSchema } = require('./src/schemas/parseJsonSchema');
-const jsonDataGenerator = require('./src/dataGenerator');
+const cleanKafka  = require('./src/kafka/cleanKafka');
+const dataGenerator = require('./src/dataGenerator');
 const fs = require('fs');
 const { program, Option } = require('commander');
 
@@ -36,6 +37,11 @@ program
             .choices(['true', 'false'])
             .default('false')
     )
+    .addOption(
+        new Option('-c, --clean <char>')
+            .choices(['true', 'false'])
+            .default('false')
+    )
     .addOption(
         new Option('-dr, --dry-run <char>', 'Dry run (no data will be produced')
             .choices(['true', 'false'])
@@ -60,6 +66,8 @@ if (!fs.existsSync(options.schema)) {
 global.debug = options.debug;
 global.recordSize = options.recordSize;
 global.wait = options.wait;
+global.clean = options.clean;
+global.dryRun = options.dryRun;
 
 if (debug === 'true') {
     console.log(options);
@@ -106,13 +114,20 @@ if (!wait) {
         process.exit();
     }
 
+    if (clean == 'true') {
+        let topics = []
+        for (table of parsedSchema){
+            topics.push(table._meta.topic)
+        }
+        await cleanKafka(options.format,topics)
+        process.exit(0);
+    }
+
     // Generate data
-    await jsonDataGenerator({
+    await dataGenerator({
         format: options.format,
         schema: parsedSchema,
-        number: options.number,
-        dryRun: options.dryRun,
-        debug: options.debug
+        number: options.number
     })
 
     await end();

package.json

@@ -30,10 +30,11 @@
     },
     "dependencies": {
         "@avro/types": "^1.0.25",
-        "@kafkajs/confluent-schema-registry": "^3.3.0",
         "@faker-js/faker": "^7.6.0",
+        "@kafkajs/confluent-schema-registry": "^3.3.0",
         "arg": "^5.0.2",
         "avro-js": "^1.11.1",
+        "axios": "^1.3.4",
         "cli-alerts": "^1.2.2",
         "cli-handle-unhandled": "^1.1.1",
         "cli-welcome": "^2.2.2",

src/dataGenerator.js

@@ -91,9 +91,7 @@ async function prepareSchema(megaRecord, topic, registry, avroSchemas) {
 module.exports = async ({
     format,
     schema,
-    number,
-    dryRun = false,
-    debug = false
+    number
 }) => {
     let payload;
     if (recordSize) {