diff --git a/CHANGES.txt b/CHANGES.txt
index a980385db79..d481c9dec00 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -127,6 +127,7 @@ Release 0.91.0 - Unreleased
    HBASE-3983  list command in shell seems broken
    HBASE-3793  HBASE-3468 Broke checkAndPut with null value (Ming Ma)
    HBASE-3995  HBASE-3946 broke TestMasterFailover
+   HBASE-3889  NPE in Distributed Log Splitting (Anirudh Todi)
 
   IMPROVEMENTS
    HBASE-3290  Max Compaction Size (Nicolas Spiegelberg via Stack)  
diff --git a/src/main/java/org/apache/hadoop/hbase/regionserver/SplitLogWorker.java b/src/main/java/org/apache/hadoop/hbase/regionserver/SplitLogWorker.java
index d39f55b8a67..843873e8fa5 100644
--- a/src/main/java/org/apache/hadoop/hbase/regionserver/SplitLogWorker.java
+++ b/src/main/java/org/apache/hadoop/hbase/regionserver/SplitLogWorker.java
@@ -136,6 +136,7 @@ public class SplitLogWorker extends ZooKeeperListener implements Runnable {
 
   @Override
   public void run() {
+   try {
     LOG.info("SplitLogWorker " + this.serverName + " starting");
     this.watcher.registerListener(this);
     int res;
@@ -162,8 +163,13 @@ public class SplitLogWorker extends ZooKeeperListener implements Runnable {
     }
 
     taskLoop();
-
-    LOG.info("SplitLogWorker " + this.serverName + " exiting");
+   } catch (Throwable t) {
+	   // only a logical error can cause here. Printing it out 
+	   // to make debugging easier
+	   LOG.error("unexpected error ", t);
+   } finally {
+	   LOG.info("SplitLogWorker " + this.serverName + " exiting");
+   }
   }
 
   /**
diff --git a/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestHLogSplit.java b/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestHLogSplit.java
index bf066732f08..de28418efea 100644
--- a/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestHLogSplit.java
+++ b/src/test/java/org/apache/hadoop/hbase/regionserver/wal/TestHLogSplit.java
@@ -152,7 +152,8 @@ public class TestHLogSplit {
    * @throws IOException
    * @see https://issues.apache.org/jira/browse/HBASE-3020
    */
-  @Test public void testRecoveredEditsPathForMeta() throws IOException {
+  @Test 
+  public void testRecoveredEditsPathForMeta() throws IOException {
     FileSystem fs = FileSystem.get(TEST_UTIL.getConfiguration());
     byte [] encoded = HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes();
     Path tdir = new Path(hbaseDir, Bytes.toString(HConstants.META_TABLE_NAME));
@@ -952,7 +953,35 @@ public class TestHLogSplit {
 
     assertEquals(true, logsAreEqual(originalLog, splitLog));
   }
+  
+  @Test
+  public void testSplitLogFileDeletedRegionDir()
+  throws IOException {
+	LOG.info("testSplitLogFileDeletedRegionDir");
+	final String REGION = "region__1";
+    regions.removeAll(regions);
+    regions.add(REGION);
 
+
+    generateHLogs(1, 10, -1);
+    FileStatus logfile = fs.listStatus(hlogDir)[0];
+    fs.initialize(fs.getUri(), conf);
+    
+    Path regiondir = new Path(tabledir, REGION);
+    LOG.info("Region directory is" + regiondir);
+    fs.delete(regiondir, true);
+    
+    HLogSplitter.splitLogFileToTemp(hbaseDir, "tmpdir", logfile, fs,
+        conf, reporter);
+    HLogSplitter.moveRecoveredEditsFromTemp("tmpdir", hbaseDir, oldLogDir,
+        logfile.getPath().toString(), conf);
+    
+    assertTrue(!fs.exists(regiondir));
+    assertTrue(true);
+  }
+
+  
+  
   @Test
   public void testSplitLogFileEmpty() throws IOException {
     LOG.info("testSplitLogFileEmpty");
@@ -1009,7 +1038,6 @@ public class TestHLogSplit {
     assertEquals(1, fs.listStatus(corruptDir).length);
   }
 
-
   private void flushToConsole(String s) {
     System.out.println(s);
     System.out.flush();