Use callback for disconnect

Author: emasab

lib/kafkajs/_kafka.js

@@ -115,10 +115,6 @@ class Producer {
         this.#internalClient.on('event.error', this.#errorCb.bind(this));
         this.#internalClient.on('event.log', console.log);
 
-        this.#internalClient.on('disconnected', (arg) => {
-            this.#state = ProducerState.DISCONNECTED;
-        });
-
         return new Promise((resolve, reject) => {
             this.#connectPromiseFunc = {resolve, reject};
             console.log("Connecting....");
@@ -127,12 +123,18 @@ class Producer {
         });
     }
 
-    disconnect() {
+    async disconnect() {
       if (this.#state >= ProducerState.DISCONNECTING) {
         return;
       }
       this.#state = ProducerState.DISCONNECTING;
-      this.#internalClient.disconnect();
+      await new Promise((resolve, reject) => {
+        const cb = (err) => {
+          err ? reject(err) : resolve();
+          this.#state = ProducerState.DISCONNECTED;
+        }
+        this.#internalClient.disconnect(5000, cb);
+      });
     }
 
     async send(sendOptions) {