From e4b51bb27d87310fd3f6a75c61a1ecda57b069bc Mon Sep 17 00:00:00 2001
From: Umesh Agashe <uagashe@cloudera.com>
Date: Wed, 4 Apr 2018 16:07:50 -0700
Subject: [PATCH] HBASE-20330 ProcedureExecutor.start() gets stuck in recover
 lease on store

rollWriter() fails after creating the file and returns false. In next iteration of while loop in recoverLease() file list is refreshed.

Signed-off-by: Appy <appy@cloudera.com>
---
 hbase-procedure/pom.xml                       |  5 +++
 .../store/wal/WALProcedureStore.java          | 13 ++++---
 .../store/wal/TestWALProcedureStore.java      | 37 +++++++++++++++++++
 3 files changed, 50 insertions(+), 5 deletions(-)

diff --git a/hbase-procedure/pom.xml b/hbase-procedure/pom.xml
index 73bc8660200..3fedda8e193 100644
--- a/hbase-procedure/pom.xml
+++ b/hbase-procedure/pom.xml
@@ -100,6 +100,11 @@
       <groupId>org.apache.hbase</groupId>
       <artifactId>hbase-metrics-api</artifactId>
     </dependency>
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <profiles>
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java
index 8581d825d5e..c5680cfa9b8 100644
--- a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java
+++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java
@@ -359,21 +359,20 @@ public class WALProcedureStore extends ProcedureStoreBase {
     lock.lock();
     try {
       LOG.trace("Starting WAL Procedure Store lease recovery");
-      FileStatus[] oldLogs = getLogFiles();
       while (isRunning()) {
+        FileStatus[] oldLogs = getLogFiles();
         // Get Log-MaxID and recover lease on old logs
         try {
           flushLogId = initOldLogs(oldLogs);
         } catch (FileNotFoundException e) {
           LOG.warn("Someone else is active and deleted logs. retrying.", e);
-          oldLogs = getLogFiles();
           continue;
         }
 
         // Create new state-log
         if (!rollWriter(flushLogId + 1)) {
           // someone else has already created this log
-          LOG.debug("Someone else has already created log " + flushLogId);
+          LOG.debug("Someone else has already created log {}. Retrying.", flushLogId);
           continue;
         }
 
@@ -1002,7 +1001,8 @@ public class WALProcedureStore extends ProcedureStoreBase {
     return true;
   }
 
-  private boolean rollWriter(final long logId) throws IOException {
+  @VisibleForTesting
+  boolean rollWriter(final long logId) throws IOException {
     assert logId > flushLogId : "logId=" + logId + " flushLogId=" + flushLogId;
     assert lock.isHeldByCurrentThread() : "expected to be the lock owner. " + lock.isLocked();
 
@@ -1072,7 +1072,10 @@ public class WALProcedureStore extends ProcedureStoreBase {
   }
 
   private void closeCurrentLogStream() {
-    if (stream == null) return;
+    if (stream == null || logs.isEmpty()) {
+      return;
+    }
+
     try {
       ProcedureWALFile log = logs.getLast();
       log.setProcIds(storeTracker.getUpdatedMinProcId(), storeTracker.getUpdatedMaxProcId());
diff --git a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestWALProcedureStore.java b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestWALProcedureStore.java
index 1929c0c7361..64cf211161e 100644
--- a/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestWALProcedureStore.java
+++ b/hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestWALProcedureStore.java
@@ -54,6 +54,9 @@ import org.junit.Before;
 import org.junit.ClassRule;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -759,6 +762,40 @@ public class TestWALProcedureStore {
     assertEquals(0, loader.getCorruptedCount());
   }
 
+  @Test
+  public void testLogFileAleadExists() throws IOException {
+    final boolean[] tested = {false};
+    WALProcedureStore mStore = Mockito.spy(procStore);
+
+    Answer<Boolean> ans = new Answer<Boolean>() {
+      @Override
+      public Boolean answer(InvocationOnMock invocationOnMock) throws Throwable {
+        long logId = ((Long) invocationOnMock.getArgument(0)).longValue();
+        switch ((int) logId) {
+          case 2:
+            // Create a file so that real rollWriter() runs into file exists condition
+            Path logFilePath = mStore.getLogFilePath(logId);
+            mStore.getFileSystem().create(logFilePath);
+            break;
+          case 3:
+            // Success only when we retry with logId 3
+            tested[0] = true;
+          default:
+            break;
+        }
+        return (Boolean) invocationOnMock.callRealMethod();
+      }
+    };
+
+    // First time Store has one log file, next id will be 2
+    Mockito.doAnswer(ans).when(mStore).rollWriter(2);
+    // next time its 3
+    Mockito.doAnswer(ans).when(mStore).rollWriter(3);
+
+    mStore.recoverLease();
+    assertTrue(tested[0]);
+  }
+
   @Test
   public void testLoadChildren() throws Exception {
     TestProcedure a = new TestProcedure(1, 0);