ZkClient doesn't handle SaslAuthenticated

[fpj]

We've set up Kafka to use ZK authentication and occasionally I've observed test failures event when everything seemed to be fine. Here is an example:

[2015-11-03 21:35:41,752] INFO zookeeper state changed (SyncConnected) (org.I0Itec.zkclient.ZkClient:704)
[2015-11-03 21:35:41,753] INFO Successfully authenticated client: authenticationID=fpj;  authorizationID=fpj. (org.apache.zookeeper.server.auth.SaslServerCallbackHandler:118)
[2015-11-03 21:35:41,753] INFO Setting authorizedID: fpj (org.apache.zookeeper.server.auth.SaslServerCallbackHandler:134)
[2015-11-03 21:35:41,753] INFO adding SASL authorization for authorizationID: fpj (org.apache.zookeeper.server.ZooKeeperServer:964)
[2015-11-03 21:35:41,754] INFO zookeeper state changed (SaslAuthenticated) (org.I0Itec.zkclient.ZkClient:704)
[2015-11-03 21:35:47,752] INFO Processed session termination for sessionid: 0x150cf46e4410002 (org.apache.zookeeper.server.PrepRequestProcessor:494)
[2015-11-03 21:35:47,755] INFO Session: 0x150cf46e4410002 closed (org.apache.zookeeper.ZooKeeper:684)
[2015-11-03 21:35:47,756] INFO Closed socket connection for client /127.0.0.1:55964 which had sessionid 0x150cf46e4410002 (org.apache.zookeeper.server.NIOServerCnxn:1007)

The client authenticates fine, but after connection timeout (one of the ZkClient constructor paramaters), it closes the session. From the Session: 0x150cf46e4410002 closed line, I can tell that the client called close() on the ZK session explicitly, so it isn't the session expiring.

Tracing the issue on ZkClient, it looks like the problem is that waitUntilConnected only waits on SyncConnected, and waitForKeeperState is missing this event. I'm thinking that this could be possible if the state condition is signaled upon receiving the SyncConnected but the ZK dispatcher thread ends up delivering SaslAuthenticated before the thread blocked on waitForKeeperState acquires the lock again.

[jzillmann]

Merged in master and released as 0.7 version!