kaisun2000 commented on a change in pull request #924: WIP: fix
ZkHelixPropertyStore loses Zookeeper notification issue
URL: https://github.com/apache/helix/pull/924#discussion_r402041626
##########
File path: helix-core/src/test/java/org/apache/helix/ZkTestHelper.java
##########
@@ -146,6 +146,70 @@ public void process(WatchedEvent event) {
LOG.info("After expiry. sessionId: " +
Long.toHexString(curZookeeper.getSessionId()));
}
+ public static void expireSharedZkClientSession(HelixZkClient client) throws
Exception {
+ final CountDownLatch waitExpireSession = new CountDownLatch(1);
+ final ZkClient zkClient = (ZkClient) client;
+
+ IZkStateListener listener = new IZkStateListener() {
+ @Override
+ public void handleStateChanged(KeeperState state) throws Exception {
+ LOG.info("IZkStateListener#handleStateChanged, state: " + state);
+ if (state == KeeperState.Expired) {
+ waitExpireSession.countDown();
+ }
+ }
+
+ @Override
+ public void handleNewSession(final String sessionId) throws Exception {
+ }
+
+ @Override
+ public void handleSessionEstablishmentError(Throwable var1) throws
Exception {
+ }
+ };
+
+ zkClient.subscribeStateChanges(listener);
+
+ ZkConnection connection = ((ZkConnection) zkClient.getConnection());
+ ZooKeeper curZookeeper = connection.getZookeeper();
+ String oldSessionId = Long.toHexString(curZookeeper.getSessionId());
+ LOG.info("Before session expiry. sessionId: " + oldSessionId + ", zk: " +
curZookeeper);
+
+ Watcher watcher = new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ LOG.info("Watcher#process, event: " + event);
+ }
+ };
+
+ final ZooKeeper dupZookeeper =
+ new ZooKeeper(connection.getServers(),
curZookeeper.getSessionTimeout(), watcher,
+ curZookeeper.getSessionId(), curZookeeper.getSessionPasswd());
+ // wait until connected, then close
+ while (dupZookeeper.getState() != States.CONNECTED) {
+ Thread.sleep(10);
+ }
+ Assert.assertEquals(dupZookeeper.getState(), States.CONNECTED,
+ "Fail to connect to zk using current session info");
+ dupZookeeper.close();
+
+ // make sure session expiry really happens
+ waitExpireSession.await();
+ zkClient.unsubscribeStateChanges(listener);
+
+ connection = (ZkConnection) zkClient.getConnection();
+ curZookeeper = connection.getZookeeper();
+
+ // wait util connected
+ while (curZookeeper.getState() != States.CONNECTED) {
Review comment:
There are many ways to do it. Let us be specific to avoid delaying this
important fix.
So the code can be something like this:
```
public void handleStateChanged(KeeperState state) throws Exception {
LOG.info("IZkStateListener#handleStateChanged, state: " + state);
if (state == KeeperState.Expired) {
waitExpireSession.countDown();
}
if (state == KeeperState.Synconnected) {
waitNewSession.countDown();
}
```
Later something like this to wait for latch?
```
boolean expiredCompleted = waitExpireSession.await(5L, TimeUnit.SECONDS);
if (!expiredCompleted) {
throw Exception(" Expiration of Session failed.");
}
boolean newSessionCompleted = waitNewSession.await(5L, TimeUnit.SECONDS);
if (!newSessionCompleted) {
throw Exception(" Expiration of Session failed.");
}
zkClient.unsubscribeStateChanges(listener);
```
Note, the wait for expiration and synconnected are both necessary. The
reason is that disconnect can give you sysconnected too.
Note, the timeout in await() is to prevent hanging here forever to block
test from proceeding.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
