moving wal marker write to store engine

apache · May 6, 2022 · a469010 · a469010
1 parent 3902b6a
commit a469010
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 9 deletions.
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java
@@ -1232,10 +1232,13 @@ private void writeCompactionWalRecord(Collection<HStoreFile> filesCompacted,
       allowedOnPath = ".*/(HStore|TestHStore).java")
   void replaceStoreFiles(Collection<HStoreFile> compactedFiles, Collection<HStoreFile> result,
     boolean writeCompactionMarker) throws IOException {
-    if (writeCompactionMarker) {
-      writeCompactionWalRecord(compactedFiles, result);
-    }
-    storeEngine.replaceStoreFiles(compactedFiles, result, () -> {
+    storeEngine.replaceStoreFiles(compactedFiles, result,
+      () -> {
+        if (writeCompactionMarker) {
+          writeCompactionWalRecord(compactedFiles, result);
+        }
+      },
+      () -> {
       synchronized (filesCompacting) {
         filesCompacting.removeAll(compactedFiles);
       }

diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreEngine.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/StoreEngine.java
@@ -34,6 +34,7 @@
 import java.util.concurrent.locks.ReadWriteLock;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
 import java.util.function.Function;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hbase.CellComparator;
@@ -407,8 +408,7 @@ private void refreshStoreFilesInternal(Collection<StoreFileInfo> newFiles) throw
     List<HStoreFile> openedFiles = openStoreFiles(toBeAddedFiles, false);
 
     // propogate the file changes to the underlying store file manager
-    replaceStoreFiles(toBeRemovedStoreFiles, openedFiles, () -> {
-    }); // won't throw an exception
+    replaceStoreFiles(toBeRemovedStoreFiles, openedFiles, () -> {}, () -> {}); // won't throw an exception
   }
 
   /**
@@ -491,9 +491,11 @@ public void addStoreFiles(Collection<HStoreFile> storeFiles,
   }
 
   public void replaceStoreFiles(Collection<HStoreFile> compactedFiles,
-    Collection<HStoreFile> newFiles, Runnable actionUnderLock) throws IOException {
+    Collection<HStoreFile> newFiles, IOExceptionRunnable walMarkerWriter,
+    Runnable actionUnderLock) throws IOException {
     storeFileTracker.replace(StoreUtils.toStoreFileInfo(compactedFiles),
       StoreUtils.toStoreFileInfo(newFiles));
+    walMarkerWriter.run();
     writeLock();
     try {
       storeFileManager.addCompactionResults(compactedFiles, newFiles);

diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStore.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestHStore.java
@@ -1017,14 +1017,14 @@ public void testRefreshStoreFilesNotChanged() throws IOException {
     // call first time after files changed
     spiedStoreEngine.refreshStoreFiles();
     assertEquals(2, this.store.getStorefilesCount());
-    verify(spiedStoreEngine, times(1)).replaceStoreFiles(any(), any(), any());
+    verify(spiedStoreEngine, times(1)).replaceStoreFiles(any(), any(), any(), any());
 
     // call second time
     spiedStoreEngine.refreshStoreFiles();
 
     // ensure that replaceStoreFiles is not called, i.e, the times does not change, if files are not
     // refreshed,
-    verify(spiedStoreEngine, times(1)).replaceStoreFiles(any(), any(), any());
+    verify(spiedStoreEngine, times(1)).replaceStoreFiles(any(), any(), any(), any());
   }
 
   private long countMemStoreScanner(StoreScanner scanner) {