Fail to close after calling zk.close() when catching error after stop zookeeper

[YunNeverMore]

I used kaka-node to connect to kafka. kafka-node use node-zookeeper-client to get connect to zookeeper to get brokers.
For abnormal testing, there will be two zookeeper connection after stop zookeeper and start zookeeper. If repeating these processes again and again, there will be so many connections even more than 100 which leads to zookeeper stuck.

zk.once('disconnected', function () { if (!zk.closed) { zk.close(); self.connect(); self.emit('zkReconnect'); } });

The problem I found is after zk.close(), the socket still does not closed and self.connect() will create another one.
So I tried to figure out the root problem in node-zookeeper-client.

`ConnectionManager.prototype.onSocketClosed = function (hasError) {
var retry = false,
errorCode,
pendingPacket;

switch (this.state) {
case STATES.CLOSING:
    errorCode = Exception.CONNECTION_LOSS;
    retry = false;
    break;
case STATES.SESSION_EXPIRED:
    errorCode = Exception.SESSION_EXPIRED;
    retry = false;
    break;
case STATES.AUTHENTICATION_FAILED:
    errorCode = Exception.AUTH_FAILED;
    retry = false;
    break;
default:
    errorCode = Exception.CONNECTION_LOSS;
    retry = true;
}

this.cleanupPendingQueue(errorCode);
this.setState(STATES.DISCONNECTED);

if (retry) {
    this.connect();
} else {
    this.setState(STATES.CLOSED);
}

};`

the socket will firstly catches error and closed event will be trigger. The 'retry' will be true. It will set state as 'disconnected' which invokes disconnected event for kaka-node to do zk.close(). Then state become 'closing' and then execute this.connect(), which will set state to 'connecting' and retry and retry.
The solution just add checking in the connect. to make sure if state == closing, just do not connect.

if (self.state === STATES.CLOSING) return;

`ConnectionManager.prototype.connect = function () {
var self = this;

if (self.state === STATES.CLOSING) return;

self.setState(STATES.CONNECTING);

self.findNextServer(function (server) {
    self.socket = net.connect(server);

    self.connectTimeoutHandler = setTimeout(
        self.onSocketConnectTimeout.bind(self),
        self.connectTimeout
    );

    // Disable the Nagle algorithm.
    self.socket.setNoDelay();

    self.socket.on('connect', self.onSocketConnected.bind(self));
    self.socket.on('data', self.onSocketData.bind(self));
    self.socket.on('drain', self.onSocketDrain.bind(self));
    self.socket.on('close', self.onSocketClosed.bind(self));
    self.socket.on('error', self.onSocketError.bind(self));
});

};`

I will send a pull request for it.
Thanks

[YunNeverMore]

Here is my pull request.
#61

Thx

[alexguan]

Thanks for the PR, I will take a look soon

[alexguan]

I think the problem exists in your test method:

zk.once('disconnected', function () {
    if (!zk.closed) {
        zk.close(); self.connect(); self.emit('zkReconnect');
    } 
});

The client.close90 is asynchronous and the self.connect should be called only after you get the disconnect event

[mariosouto]

@YunNeverMore I'm having this same issue, did you figure out on how to solve it?

@alexguan How can we now if zk.close() successfully close the connection? I was checking if maybe it returned a Promise or something similar but it doesn't. Also, i believe this snippet of code is from the kafka-node repo