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 <[email protected]>

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>

hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/store/wal/WALProcedureStore.java

@@ -359,21 +359,20 @@ public void recoverLease() throws IOException {
     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 @@ private boolean rollWriter() throws IOException {
     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 @@ private boolean rollWriter(final long logId) throws IOException {
   }
 
   private void closeCurrentLogStream() {
-    if (stream == null) return;
+    if (stream == null || logs.isEmpty()) {
+      return;
+    }
+
     try {
       ProcedureWALFile log = logs.getLast();
       log.setProcIds(storeTracker.getUpdatedMinProcId(), storeTracker.getUpdatedMaxProcId());

hbase-procedure/src/test/java/org/apache/hadoop/hbase/procedure2/store/wal/TestWALProcedureStore.java

@@ -54,6 +54,9 @@
 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 void recoverFileLease(FileSystem fs, Path path) throws IOException {
     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);