Index: src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java =================================================================== --- src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java (revision 1196325) +++ src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSourceManager.java (working copy) @@ -210,7 +210,14 @@ if (this.latestPath != null) { String name = this.latestPath.getName(); this.hlogsById.get(id).add(name); - this.zkHelper.addLogToList(name, src.getPeerClusterZnode()); + try { + this.zkHelper.addLogToList(name, src.getPeerClusterZnode()); + } catch (KeeperException ke) { + String message = "Cannot add log to zk for" + + " replication when creating a new source"; + stopper.stop(message); + throw new IOException(message, ke); + } src.enqueueLog(this.latestPath); } } @@ -247,7 +254,7 @@ return this.sources; } - void logRolled(Path newLog) { + void logRolled(Path newLog) throws IOException { if (!this.replicating.get()) { LOG.warn("Replication stopped, won't add new log"); return; @@ -256,7 +263,11 @@ synchronized (this.hlogsById) { String name = newLog.getName(); for (ReplicationSourceInterface source : this.sources) { - this.zkHelper.addLogToList(name, source.getPeerClusterZnode()); + try { + this.zkHelper.addLogToList(name, source.getPeerClusterZnode()); + } catch (KeeperException ke) { + throw new IOException("Cannot add log to zk for replication", ke); + } } for (SortedSet hlogs : this.hlogsById.values()) { if (this.sources.isEmpty()) { Index: src/main/java/org/apache/hadoop/hbase/replication/ReplicationZookeeper.java =================================================================== --- src/main/java/org/apache/hadoop/hbase/replication/ReplicationZookeeper.java (revision 1196325) +++ src/main/java/org/apache/hadoop/hbase/replication/ReplicationZookeeper.java (working copy) @@ -432,14 +432,11 @@ * @param filename name of the hlog's znode * @param peerId name of the cluster's znode */ - public void addLogToList(String filename, String peerId) { - try { - String znode = ZKUtil.joinZNode(this.rsServerNameZnode, peerId); - znode = ZKUtil.joinZNode(znode, filename); - ZKUtil.createWithParents(this.zookeeper, znode); - } catch (KeeperException e) { - this.abortable.abort("Failed add log to list", e); - } + public void addLogToList(String filename, String peerId) + throws KeeperException { + String znode = ZKUtil.joinZNode(this.rsServerNameZnode, peerId); + znode = ZKUtil.joinZNode(znode, filename); + ZKUtil.createWithParents(this.zookeeper, znode); } /**