HBASE-30341: Fix FSHLog WAL lockup on uncaught exception in publishSyncOnRingBuffer

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

@@ -1087,7 +1087,7 @@ public void updateStore(byte[] encodedRegionName, byte[] familyName, Long sequen
     sequenceIdAccounting.updateStore(encodedRegionName, familyName, sequenceid, onlyIfGreater);
   }
 
-  protected final SyncFuture getSyncFuture(long sequence, boolean forceSync) {
+  protected SyncFuture getSyncFuture(long sequence, boolean forceSync) {
     return syncFutureCache.getIfPresentOrNew().reset(sequence, forceSync);
   }
 

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

@@ -794,8 +794,11 @@ private SyncFuture publishSyncOnRingBuffer(boolean forceSync) {
 
   protected SyncFuture publishSyncOnRingBuffer(long sequence, boolean forceSync) {
     // here we use ring buffer sequence as transaction id
-    SyncFuture syncFuture = getSyncFuture(sequence, forceSync);
+    // getSyncFuture must stay inside the try: the sequence is already claimed, so we must publish
+    // it even if this throws, else the consumer wedges.
+    SyncFuture syncFuture = null;
     try {
+      syncFuture = getSyncFuture(sequence, forceSync);
       RingBufferTruck truck = this.disruptor.getRingBuffer().get(sequence);
       truck.load(syncFuture);
     } finally {
@@ -1079,11 +1082,13 @@ public void onEvent(final RingBufferTruck truck, final long sequence, boolean en
           } finally {
             entry.release();
           }
+        } else if (truck.type() == RingBufferTruck.Type.EMPTY) {
+          // publishSyncOnRingBuffer claimed the sequence but threw before loading the truck.
+          LOG.warn("Empty RingBufferTruck at sequence {}", sequence);
+          return;
         } else {
-          // What is this if not an append or sync. Fail all up to this!!!
           cleanupOutstandingSyncsOnException(sequence,
-            new IllegalStateException("Neither append nor sync"));
-          // Return to keep processing.
+            new IllegalStateException("Unexpected truck type: " + truck.type()));
           return;
         }
 

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

@@ -21,6 +21,8 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import org.apache.hbase.thirdparty.com.google.common.cache.Cache;
 import org.apache.hbase.thirdparty.com.google.common.cache.CacheBuilder;
@@ -45,6 +47,8 @@
 @InterfaceAudience.Private
 public final class SyncFutureCache {
 
+  private static final Logger LOG = LoggerFactory.getLogger(SyncFutureCache.class);
+
   private static final long SYNC_FUTURE_INVALIDATION_TIMEOUT_MINS = 2;
 
   private final Cache<Thread, SyncFuture> syncFutureCache;
@@ -57,9 +61,15 @@ public SyncFutureCache(final Configuration conf) {
   }
 
   public SyncFuture getIfPresentOrNew() {
-    // Invalidate the entry if a mapping exists. We do not want it to be reused at the same time.
-    SyncFuture future = syncFutureCache.asMap().remove(Thread.currentThread());
-    return (future == null) ? new SyncFuture() : future;
+    // The cache is only an allocation optimisation; never let it fail a write.
+    try {
+      // Invalidate the entry if a mapping exists. We do not want it to be reused at the same time.
+      SyncFuture future = syncFutureCache.asMap().remove(Thread.currentThread());
+      return (future == null) ? new SyncFuture() : future;
+    } catch (RuntimeException e) {
+      LOG.warn("SyncFutureCache lookup failed; falling back to a new SyncFuture", e);
+      return new SyncFuture();
+    }
   }
 
   /**

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

@@ -20,6 +20,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -422,4 +423,52 @@ public void testGetPipelineDoesNotReturnNullWhenUnderlyingStreamerHasNone() thro
         "Should normalize a null underlying pipeline to an empty array");
     }
   }
+
+  /**
+   * HBASE-30341: an exception out of {@code getSyncFuture} used to leave the claimed ring buffer
+   * sequence unpublished, wedging the WAL. If it regresses, this test times out on the final sync.
+   */
+  @Test
+  public void testGetSyncFutureThrowDoesNotWedgeWAL() throws IOException {
+    class DodgyFSHLog extends FSHLog {
+      volatile boolean throwException = false;
+
+      DodgyFSHLog(FileSystem fs, Path rootDir, String logDir, Configuration conf)
+        throws IOException {
+        super(fs, rootDir, logDir, conf);
+      }
+
+      @Override
+      protected SyncFuture getSyncFuture(long sequence, boolean forceSync) {
+        if (throwException) {
+          throw new RuntimeException("FAKE! getSyncFuture blew up");
+        }
+        return super.getSyncFuture(sequence, forceSync);
+      }
+    }
+
+    TableDescriptor td = TableDescriptorBuilder.newBuilder(TableName.valueOf(name))
+      .setColumnFamily(ColumnFamilyDescriptorBuilder.of("row")).build();
+    RegionInfo ri = RegionInfoBuilder.newBuilder(td.getTableName()).build();
+    MultiVersionConcurrencyControl mvcc = new MultiVersionConcurrencyControl();
+    NavigableMap<byte[], Integer> scopes = new TreeMap<>(Bytes.BYTES_COMPARATOR);
+    for (byte[] fam : td.getColumnFamilyNames()) {
+      scopes.put(fam, 0);
+    }
+
+    DodgyFSHLog wal = new DodgyFSHLog(FS, CommonFSUtils.getWALRootDir(CONF), DIR.toString(), CONF);
+    wal.init();
+    try {
+      addEdits(wal, ri, td, 1, mvcc, scopes, "row");
+
+      wal.throwException = true;
+      assertThrows(Exception.class, () -> addEdits(wal, ri, td, 1, mvcc, scopes, "row"));
+
+      // WAL must still be usable; pre-fix this hangs forever on the leaked sequence.
+      wal.throwException = false;
+      addEdits(wal, ri, td, 1, mvcc, scopes, "row");
+    } finally {
+      wal.close();
+    }
+  }
 }

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

@@ -21,14 +21,20 @@
 import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNotSame;
 
+import java.lang.reflect.Field;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentMap;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.testclassification.RegionServerTests;
 import org.apache.hadoop.hbase.testclassification.SmallTests;
 import org.junit.jupiter.api.Tag;
 import org.junit.jupiter.api.Test;
 
+import org.apache.hbase.thirdparty.com.google.common.cache.Cache;
+import org.apache.hbase.thirdparty.com.google.common.cache.CacheBuilder;
+import org.apache.hbase.thirdparty.com.google.common.cache.ForwardingCache;
+
 @Tag(RegionServerTests.TAG)
 @Tag(SmallTests.TAG)
 public class TestSyncFutureCache {
@@ -62,4 +68,28 @@ public void testSyncFutureCacheLifeCycle() throws Exception {
       cache.clear();
     }
   }
+
+  @Test
+  public void testFallsBackToNewSyncFutureWhenCacheThrows() throws Exception {
+    SyncFutureCache cache = new SyncFutureCache(HBaseConfiguration.create());
+
+    final Cache<Thread, SyncFuture> delegate = CacheBuilder.newBuilder().build();
+    Cache<Thread, SyncFuture> throwing = new ForwardingCache<Thread, SyncFuture>() {
+      @Override
+      protected Cache<Thread, SyncFuture> delegate() {
+        return delegate;
+      }
+
+      @Override
+      public ConcurrentMap<Thread, SyncFuture> asMap() {
+        throw new NullPointerException("boom");
+      }
+    };
+
+    Field field = SyncFutureCache.class.getDeclaredField("syncFutureCache");
+    field.setAccessible(true);
+    field.set(cache, throwing);
+
+    assertNotNull(cache.getIfPresentOrNew());
+  }
 }