diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java index f18b8ba..29947c7 100644 --- a/hbase-client/src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java +++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/zookeeper/ZooKeeperWatcher.java @@ -207,7 +207,7 @@ public class ZooKeeperWatcher implements Watcher, Abortable, Closeable { } catch(KeeperException.NoNodeException nne) { return; } catch(InterruptedException ie) { - interruptedException(ie); + interruptedExceptionNoThrow(ie, false); } catch (IOException|KeeperException e) { LOG.warn("Received exception while checking and setting zookeeper ACLs", e); } @@ -587,20 +587,26 @@ public class ZooKeeperWatcher implements Watcher, Abortable, Closeable { /** * Handles InterruptedExceptions in client calls. - *
- * This may be temporary but for now this gives one place to deal with these. - *
- * TODO: Currently, this method does nothing. - * Is this ever expected to happen? Do we abort or can we let it run? - * Maybe this should be logged as WARN? It shouldn't happen? - *
- * @param ie + * @param ie the InterruptedException instance thrown + * @throws KeeperException the exception to throw, transformed from the InterruptedException + */ + public void interruptedException(InterruptedException ie) throws KeeperException { + interruptedExceptionNoThrow(ie, true); + // Throw a system error exception to let upper level handle it + throw new KeeperException.SystemErrorException(); + } + + /** + * Log the InterruptedException and interrupt current thread + * @param ie The IterruptedException to log + * @param throwLater Whether we will throw the exception latter */ - public void interruptedException(InterruptedException ie) { - LOG.debug(prefix("Received InterruptedException, doing nothing here"), ie); + public void interruptedExceptionNoThrow(InterruptedException ie, boolean throwLater) { + LOG.debug(prefix("Received InterruptedException, will interrupt current thread" + + (throwLater ? " and rethrow a SystemErrorException" : "")), + ie); // At least preserver interrupt. Thread.currentThread().interrupt(); - // no-op } /** diff --git a/hbase-client/src/test/java/org/apache/hadoop/hbase/zookeeper/TestZKUtil.java b/hbase-client/src/test/java/org/apache/hadoop/hbase/zookeeper/TestZKUtil.java index 076569b..29beb78 100644 --- a/hbase-client/src/test/java/org/apache/hadoop/hbase/zookeeper/TestZKUtil.java +++ b/hbase-client/src/test/java/org/apache/hadoop/hbase/zookeeper/TestZKUtil.java @@ -26,6 +26,7 @@ import org.apache.hadoop.hbase.HBaseConfiguration; import org.apache.hadoop.hbase.ZooKeeperConnectionException; import org.apache.hadoop.hbase.security.Superusers; import org.apache.hadoop.hbase.testclassification.SmallTests; +import org.apache.zookeeper.KeeperException; import org.apache.zookeeper.ZooDefs.Ids; import org.apache.zookeeper.ZooDefs.Perms; import org.apache.zookeeper.data.ACL; @@ -33,6 +34,7 @@ import org.apache.zookeeper.data.Id; import org.junit.Assert; import org.junit.Test; import org.junit.experimental.categories.Category; +import org.mockito.Mockito; /** * @@ -77,4 +79,25 @@ public class TestZKUtil { Assert.assertTrue(aclList.contains(new ACL(Perms.ALL, new Id("sasl", "user2")))); Assert.assertTrue(aclList.contains(new ACL(Perms.ALL, new Id("sasl", "user3")))); } + + @Test + public void testInterruptedDuringAction() + throws ZooKeeperConnectionException, IOException, KeeperException, InterruptedException { + final RecoverableZooKeeper recoverableZk = Mockito.mock(RecoverableZooKeeper.class); + ZooKeeperWatcher zkw = new ZooKeeperWatcher(HBaseConfiguration.create(), "unittest", null) { + @Override + public RecoverableZooKeeper getRecoverableZooKeeper() { + return recoverableZk; + } + }; + Mockito.doThrow(new InterruptedException()).when(recoverableZk) + .getChildren(zkw.znodePaths.baseZNode, null); + try { + ZKUtil.listChildrenNoWatch(zkw, zkw.znodePaths.baseZNode); + } catch (KeeperException.SystemErrorException e) { + // expected + return; + } + Assert.fail("Should have thrown KeeperException but not"); + } }