bucket notification - move kafka options into object. Also add missing call to destroy() (#8889)

Signed-off-by: Amit Prinz Setter <alphaprinz@gmail.com>

Author: alphaprinz

docs/NooBaaNonContainerized/NooBaaCLI.md

@@ -569,6 +569,16 @@ noobaa-cli connection add --from_file
    - Type: Object
    - Description: An object given as options to node http(s) request. If "auth" field is specified, it's value is encrypted.
 
+- `kafka_options_object`
+   - Type: Object
+   - Description: An object given as options to kafka client.
+     Options are listed in https://github.com/edenhill/librdkafka/blob/v2.8.0/CONFIGURATION.md.
+     Specifically, 'metadata.broker.list' is used to specify the external kafka server.
+
+- `topic`
+   - Type: String
+   - Description - Topic for kafka messages.
+
 - `from_file`
     - Type: String
     - Description: Path to a JSON file which includes connection properties. When using `from_file` flag the connection details must only appear inside the options JSON file. See example below.

src/manage_nsfs/manage_nsfs_constants.js

@@ -89,7 +89,7 @@ const VALID_OPTIONS_UPGRADE = {
 const VALID_OPTIONS_NOTIFICATION = {};
 
 const VALID_OPTIONS_CONNECTION = {
-    'add': new Set(['name', 'notification_protocol', 'agent_request_object', 'request_options_object', FROM_FILE, ...CLI_MUTUAL_OPTIONS]),
+    'add': new Set(['name', 'notification_protocol', 'agent_request_object', 'request_options_object', 'topic', 'kafka_options_object', FROM_FILE, ...CLI_MUTUAL_OPTIONS]),
     'update': new Set(['name', 'key', 'value', 'remove_key', ...CLI_MUTUAL_OPTIONS]),
     'delete': new Set(['name', ...CLI_MUTUAL_OPTIONS]),
     'list': new Set(CLI_MUTUAL_OPTIONS),
@@ -163,6 +163,8 @@ const OPTION_TYPE = {
     notification_protocol: 'string',
     agent_request_object: 'string',
     request_options_object: 'string',
+    kafka_options_object: 'string',
+    topic: 'string',
     decrypt: 'boolean',
     key: 'string',
     value: 'string',

src/test/unit_tests/jest_tests/test_nc_nsfs_connection_cli.test.js

@@ -64,11 +64,28 @@ describe('manage nsfs cli connection flow', () => {
             conn_options.name = "fromcli";
             const res = await exec_manage_cli(TYPES.CONNECTION, ACTIONS.ADD, conn_options);
             const res_json = JSON.parse(res.trim());
-            expect(_.isEqual(new Set(Object.keys(res_json.response)), new Set(['reply', 'code']))).toBe(true);
+            expect(_.isEqual(new Set(Object.keys(res_json.response)), new Set(['reply', 'code', 'message']))).toBe(true);
             const connection = await config_fs.get_connection_by_name(conn_options.name);
             assert_connection(connection, conn_options, true);
         }, timeout);
 
+        it('cli create connection from cli - kafka', async () => {
+            const conn_options = {
+                config_root,
+                notification_protocol: 'kafka',
+                topic: 'mytopic',
+                kafka_options_object: {
+                    'metadata.broker.list': 'localhost:9092'
+                }
+            };
+            conn_options.name = "fromcli";
+            const res = await exec_manage_cli(TYPES.CONNECTION, ACTIONS.ADD, conn_options);
+            const res_json = JSON.parse(res.trim());
+            expect(_.isEqual(new Set(Object.keys(res_json.response)), new Set(['reply', 'code', 'message']))).toBe(true);
+            const connection = await config_fs.get_connection_by_name(conn_options.name);
+            assert_connection(connection, conn_options, false, true);
+        }, timeout);
+
         it('cli delete connection ', async () => {
             await exec_manage_cli(TYPES.CONNECTION, ACTIONS.DELETE, {config_root, name: defaults.name});
             expect(fs.readdirSync(config_fs.connections_dir_path).filter(file => file.endsWith(".json")).length).toEqual(0);
@@ -131,15 +148,21 @@ describe('manage nsfs cli connection flow', () => {
  * @param {object} connection actual
  * @param {object} connection_options expected
  * @param {boolean} is_encrypted whether connection's auth field is encrypted
+ * @param {boolean} is_kafka check for kafka fields (defualt is for http fields)
  */
-function assert_connection(connection, connection_options, is_encrypted) {
+function assert_connection(connection, connection_options, is_encrypted, is_kafka) {
     expect(connection.name).toEqual(connection_options.name);
     expect(connection.notification_protocol).toEqual(connection_options.notification_protocol);
-    expect(connection.agent_request_object).toStrictEqual(connection_options.agent_request_object);
-    if (is_encrypted) {
-        expect(connection.request_options_object).not.toStrictEqual(connection_options.request_options_object);
+    if (is_kafka) {
+        expect(connection.topic).toStrictEqual(connection_options.topic);
+        expect(connection.kafka_options_object).toStrictEqual(connection_options.kafka_options_object);
     } else {
-        expect(connection.request_options_object).toStrictEqual(connection_options.request_options_object);
+        expect(connection.agent_request_object).toStrictEqual(connection_options.agent_request_object);
+        if (is_encrypted) {
+            expect(connection.request_options_object).not.toStrictEqual(connection_options.request_options_object);
+        } else {
+            expect(connection.request_options_object).toStrictEqual(connection_options.request_options_object);
+        }
     }
 }
 

src/util/notifications_util.js

@@ -104,6 +104,9 @@ class Notificator {
             } finally {
                 await log.close();
                 this.notif_to_connect.clear();
+                for (const conn of this.connect_str_to_connection.values()) {
+                    conn.destroy();
+                }
             }
         }
     }
@@ -255,13 +258,7 @@ class KafkaNotificator {
     }
 
     async connect() {
-        //kafka client doens't like options it's not familiar with
-        //so delete them before connecting
-        const connect_for_kafka = structuredClone(this.connect_obj);
-        delete connect_for_kafka.topic;
-        delete connect_for_kafka.notification_protocol;
-        delete connect_for_kafka.name;
-        this.connection = new Kafka.HighLevelProducer(connect_for_kafka);
+        this.connection = new Kafka.HighLevelProducer(this.connect_obj.kafka_options_object);
         await new Promise((res, rej) => {
             this.connection.on('ready', () => {
                 res();