HBASE-28569: fix race condition during WAL splitting leading to corrupt recovered.edits

If an exception happens in the call to finishWriterThreads in the
org.apache.hadoop.hbase.wal.RecoveredEditsOutputSink.close method,
the call to closeWriters should not execute, as it may lead to a race condition
that leads to file corruption if the regionserver aborts. The execution of
closeWriters in this case would write the trailer in parallel with writer threads,
causing corruption, and then the corrupt file would get renamed and finalized
when it should not be. This corruption causes problems when the region is then
to be assigned.
To fix this, when finishWriterThreads throws an exception or is not successful,
the corrupt files should not be renamed and finalized.

Author: ciacono

hbase-server/src/main/java/org/apache/hadoop/hbase/wal/AbstractRecoveredEditsOutputSink.java

@@ -73,33 +73,27 @@ protected RecoveredEditsWriter createRecoveredEditsWriter(TableName tableName, b
     return new RecoveredEditsWriter(region, regionEditsPath, w, seqId);
   }
 
-  protected Path closeRecoveredEditsWriter(RecoveredEditsWriter editsWriter,
+  /**
+   * abortRecoveredEditsWriter closes the editsWriter, but does not rename and finalize the
+   * recovered edits WAL files. Please see HBASE-28569.
+   */
+  protected void abortRecoveredEditsWriter(RecoveredEditsWriter editsWriter,
     List<IOException> thrown) throws IOException {
-    try {
-      editsWriter.writer.close();
-    } catch (IOException ioe) {
-      final String errorMsg = "Could not close recovered edits at " + editsWriter.path;
-      LOG.error(errorMsg, ioe);
-      updateStatusWithMsg(errorMsg);
-      thrown.add(ioe);
+    closeRecoveredEditsWriter(editsWriter, thrown);
+    if (editsWriter.editsWritten == 0) {
+      // just remove the empty recovered.edits file
+      removeRecoveredEditsFile(editsWriter);
+    }
+  }
+
+  protected Path closeRecoveredEditsWriterAndFinalizeEdits(RecoveredEditsWriter editsWriter,
+    List<IOException> thrown) throws IOException {
+    if (!closeRecoveredEditsWriter(editsWriter, thrown)) {
       return null;
     }
-    final String msg = "Closed recovered edits writer path=" + editsWriter.path + " (wrote "
-      + editsWriter.editsWritten + " edits, skipped " + editsWriter.editsSkipped + " edits in "
-      + (editsWriter.nanosSpent / 1000 / 1000) + " ms)";
-    LOG.info(msg);
-    updateStatusWithMsg(msg);
     if (editsWriter.editsWritten == 0) {
       // just remove the empty recovered.edits file
-      if (
-        walSplitter.walFS.exists(editsWriter.path)
-          && !walSplitter.walFS.delete(editsWriter.path, false)
-      ) {
-        final String errorMsg = "Failed deleting empty " + editsWriter.path;
-        LOG.warn(errorMsg);
-        updateStatusWithMsg(errorMsg);
-        throw new IOException("Failed deleting empty  " + editsWriter.path);
-      }
+      removeRecoveredEditsFile(editsWriter);
       return null;
     }
 
@@ -133,6 +127,37 @@ protected Path closeRecoveredEditsWriter(RecoveredEditsWriter editsWriter,
     return dst;
   }
 
+  private boolean closeRecoveredEditsWriter(RecoveredEditsWriter editsWriter,
+    List<IOException> thrown) {
+    try {
+      editsWriter.writer.close();
+    } catch (IOException ioe) {
+      final String errorMsg = "Could not close recovered edits at " + editsWriter.path;
+      LOG.error(errorMsg, ioe);
+      updateStatusWithMsg(errorMsg);
+      thrown.add(ioe);
+      return false;
+    }
+    final String msg = "Closed recovered edits writer path=" + editsWriter.path + " (wrote "
+      + editsWriter.editsWritten + " edits, skipped " + editsWriter.editsSkipped + " edits in "
+      + (editsWriter.nanosSpent / 1000 / 1000) + " ms)";
+    LOG.info(msg);
+    updateStatusWithMsg(msg);
+    return true;
+  }
+
+  private void removeRecoveredEditsFile(RecoveredEditsWriter editsWriter) throws IOException {
+    if (
+      walSplitter.walFS.exists(editsWriter.path)
+        && !walSplitter.walFS.delete(editsWriter.path, false)
+    ) {
+      final String errorMsg = "Failed deleting empty " + editsWriter.path;
+      LOG.warn(errorMsg);
+      updateStatusWithMsg(errorMsg);
+      throw new IOException("Failed deleting empty  " + editsWriter.path);
+    }
+  }
+
   @Override
   public boolean keepRegionEvent(WAL.Entry entry) {
     ArrayList<Cell> cells = entry.getEdit().getCells();

hbase-server/src/main/java/org/apache/hadoop/hbase/wal/BoundedRecoveredEditsOutputSink.java

@@ -70,7 +70,7 @@ public void append(EntryBuffers.RegionEntryBuffer buffer) throws IOException {
       regionEditsWrittenMap.compute(Bytes.toString(buffer.encodedRegionName),
         (k, v) -> v == null ? writer.editsWritten : v + writer.editsWritten);
       List<IOException> thrown = new ArrayList<>();
-      Path dst = closeRecoveredEditsWriter(writer, thrown);
+      Path dst = closeRecoveredEditsWriterAndFinalizeEdits(writer, thrown);
       splits.add(dst);
       openingWritersNum.decrementAndGet();
       if (!thrown.isEmpty()) {

hbase-server/src/main/java/org/apache/hadoop/hbase/wal/RecoveredEditsOutputSink.java

@@ -89,24 +89,39 @@ private RecoveredEditsWriter getRecoveredEditsWriter(TableName tableName, byte[]
 
   @Override
   public List<Path> close() throws IOException {
-    boolean isSuccessful = true;
+    boolean isSuccessful;
     try {
       isSuccessful = finishWriterThreads();
-    } finally {
-      isSuccessful &= closeWriters();
+    } catch (IOException e) {
+      closeWriters(false);
+      throw e;
+    }
+    if (!isSuccessful) {
+      // Even if an exception is not thrown, finishWriterThreads() not being successful is an
+      // error case where the WAL files should not be finalized.
+      closeWriters(false);
+      return null;
     }
+    isSuccessful = closeWriters(true);
     return isSuccessful ? splits : null;
   }
 
   /**
-   * Close all of the output streams.
+   * Close all the output streams.
+   * @param finalizeEdits true in the successful close case, false when we don't want to rename and
+   *                      finalize the temporary, possibly corrupted WAL files, such as when there
+   *                      was a previous failure or exception. Please see HBASE-28569.
    * @return true when there is no error.
    */
-  private boolean closeWriters() throws IOException {
+  boolean closeWriters(boolean finalizeEdits) throws IOException {
     List<IOException> thrown = Lists.newArrayList();
     for (RecoveredEditsWriter writer : writers.values()) {
       closeCompletionService.submit(() -> {
-        Path dst = closeRecoveredEditsWriter(writer, thrown);
+        if (!finalizeEdits) {
+          abortRecoveredEditsWriter(writer, thrown);
+          return null;
+        }
+        Path dst = closeRecoveredEditsWriterAndFinalizeEdits(writer, thrown);
         LOG.trace("Closed {}", dst);
         splits.add(dst);
         return null;

hbase-server/src/test/java/org/apache/hadoop/hbase/wal/TestRecoveredEditsOutputSink.java

@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hbase.wal;
+
+import static org.junit.Assert.assertThrows;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.HBaseClassTestRule;
+import org.apache.hadoop.hbase.HBaseTestingUtil;
+import org.apache.hadoop.hbase.testclassification.RegionServerTests;
+import org.apache.hadoop.hbase.testclassification.SmallTests;
+import org.apache.hadoop.hbase.util.CommonFSUtils;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.ClassRule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.mockito.Mockito;
+
+@Category({ RegionServerTests.class, SmallTests.class })
+public class TestRecoveredEditsOutputSink {
+
+  @ClassRule
+  public static final HBaseClassTestRule CLASS_RULE =
+    HBaseClassTestRule.forClass(TestRecoveredEditsOutputSink.class);
+
+  private static WALFactory wals;
+  private static FileSystem fs;
+  private static Path rootDir;
+  private final static HBaseTestingUtil TEST_UTIL = new HBaseTestingUtil();
+
+  private static RecoveredEditsOutputSink outputSink;
+
+  @BeforeClass
+  public static void setUpBeforeClass() throws Exception {
+    Configuration conf = TEST_UTIL.getConfiguration();
+    conf.set(WALFactory.WAL_PROVIDER, "filesystem");
+    rootDir = TEST_UTIL.createRootDir();
+    fs = CommonFSUtils.getRootDirFileSystem(conf);
+    wals = new WALFactory(conf, "testRecoveredEditsOutputSinkWALFactory");
+    WALSplitter splitter = new WALSplitter(wals, conf, rootDir, fs, rootDir, fs);
+    WALSplitter.PipelineController pipelineController = new WALSplitter.PipelineController();
+    EntryBuffers sink = new EntryBuffers(pipelineController, 1024 * 1024);
+    outputSink = new RecoveredEditsOutputSink(splitter, pipelineController, sink, 3);
+  }
+
+  @AfterClass
+  public static void tearDownAfterClass() throws Exception {
+    wals.close();
+    fs.delete(rootDir, true);
+  }
+
+  @Test
+  public void testCloseSuccess() throws IOException {
+    RecoveredEditsOutputSink spyOutputSink = Mockito.spy(outputSink);
+    spyOutputSink.close();
+    Mockito.verify(spyOutputSink, Mockito.times(1)).finishWriterThreads();
+    Mockito.verify(spyOutputSink, Mockito.times(1)).closeWriters(true);
+  }
+
+  /**
+   * When a WAL split is interrupted (ex. by a RegionServer abort), the thread join in
+   * finishWriterThreads() will get interrupted, rethrowing the exception without stopping the
+   * writer threads. Test to ensure that when this happens, RecoveredEditsOutputSink.close() does
+   * not rename the recoveredEdits WAL files as this can cause corruption. Please see HBASE-28569.
+   * However, the writers must still be closed.
+   */
+  @Test
+  public void testCloseWALSplitInterrupted() throws IOException {
+    RecoveredEditsOutputSink spyOutputSink = Mockito.spy(outputSink);
+    // The race condition will lead to an InterruptedException to be caught by finishWriterThreads()
+    // which is then rethrown as an InterruptedIOException.
+    Mockito.doThrow(new InterruptedIOException()).when(spyOutputSink).finishWriterThreads();
+    assertThrows(InterruptedIOException.class, spyOutputSink::close);
+    Mockito.verify(spyOutputSink, Mockito.times(1)).finishWriterThreads();
+    Mockito.verify(spyOutputSink, Mockito.times(1)).closeWriters(false);
+  }
+
+  /**
+   * When finishWriterThreads fails but does not throw an exception, ensure the writers are handled
+   * like in the exception case - the writers are closed but the recoveredEdits WAL files are not
+   * renamed.
+   */
+  @Test
+  public void testCloseWALFinishWriterThreadsFailed() throws IOException {
+    RecoveredEditsOutputSink spyOutputSink = Mockito.spy(outputSink);
+    Mockito.doReturn(false).when(spyOutputSink).finishWriterThreads();
+    spyOutputSink.close();
+    Mockito.verify(spyOutputSink, Mockito.times(1)).finishWriterThreads();
+    Mockito.verify(spyOutputSink, Mockito.times(1)).closeWriters(false);
+  }
+}