HBASE-24822 Add a command to support archive the earliest log file ma…

hbase-client/src/main/java/org/apache/hadoop/hbase/client/Admin.java

@@ -1322,6 +1322,14 @@ Future<Boolean> abortProcedureAsync(long procId, boolean mayInterruptIfRunning)
    */
   void rollWALWriter(ServerName serverName) throws IOException, FailedLogCloseException;
 
+  /**
+   * Archive the earliest log file.
+   *
+   * @param serverName The servername of the regionserver.
+   * @throws IOException if a remote or network exception occurs
+   */
+  void archiveWAL(ServerName serverName) throws IOException;
+
   /**
    * Helper that delegates to getClusterMetrics().getMasterCoprocessorNames().
    * @return an array of master coprocessors

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AdminOverAsyncAdmin.java

@@ -588,6 +588,11 @@ public void rollWALWriter(ServerName serverName) throws IOException, FailedLogCl
     get(admin.rollWALWriter(serverName));
   }
 
+  @Override
+  public void archiveWAL(ServerName serverName) throws IOException {
+    get(admin.archiveWAL(serverName));
+  }
+
   @Override
   public CompactionState getCompactionState(TableName tableName) throws IOException {
     return get(admin.getCompactionState(tableName));

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java

@@ -1127,6 +1127,13 @@ default CompletableFuture<Integer> getMasterInfoPort() {
    */
   CompletableFuture<Void> rollWALWriter(ServerName serverName);
 
+  /**
+   * Archive the earliest log file.
+   *
+   * @param serverName The servername of the region server.
+   */
+  CompletableFuture<Void> archiveWAL(ServerName serverName);
+
   /**
    * Clear compacting queues on a region server.
    * @param serverName

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java

@@ -635,6 +635,11 @@ public CompletableFuture<Void> rollWALWriter(ServerName serverName) {
     return wrap(rawAdmin.rollWALWriter(serverName));
   }
 
+  @Override
+  public CompletableFuture<Void> archiveWAL(ServerName serverName) {
+    return wrap(rawAdmin.archiveWAL(serverName));
+  }
+
   @Override
   public CompletableFuture<Void> clearCompactionQueues(ServerName serverName, Set<String> queues) {
     return wrap(rawAdmin.clearCompactionQueues(serverName, queues));

hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncHBaseAdmin.java

@@ -121,6 +121,8 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AccessControlProtos.RevokeRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AccessControlProtos.RevokeResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ArchiveWALRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ArchiveWALResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearRegionBlockCacheRequest;
@@ -2913,6 +2915,17 @@ public CompletableFuture<Void> rollWALWriter(ServerName serverName) {
         .serverName(serverName).call();
   }
 
+  @Override
+  public CompletableFuture<Void> archiveWAL(ServerName serverName) {
+    return this
+      .<Void> newAdminCaller()
+      .action(
+        (controller, stub) -> this.<ArchiveWALRequest, ArchiveWALResponse, Void> adminCall(
+          controller, stub, RequestConverter.buildArchiveWALRequest(),
+          (s, c, req, done) -> s.archiveWAL(controller, req, done), resp -> null))
+      .serverName(serverName).call();
+  }
+
   @Override
   public CompletableFuture<Void> clearCompactionQueues(ServerName serverName, Set<String> queues) {
     return this

hbase-client/src/main/java/org/apache/hadoop/hbase/shaded/protobuf/RequestConverter.java

@@ -72,6 +72,7 @@
 import org.apache.hbase.thirdparty.com.google.common.collect.Sets;
 import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
 
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ArchiveWALRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearRegionBlockCacheRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearSlowLogResponseRequest;
@@ -883,6 +884,20 @@ public static RollWALWriterRequest buildRollWALWriterRequest() {
     return ROLL_WAL_WRITER_REQUEST;
   }
 
+  /**
+   * @see #buildArchiveWALRequest()
+   */
+  private static ArchiveWALRequest ARCHIVE_WAL_REQUEST = ArchiveWALRequest.newBuilder()
+    .build();
+
+  /**
+   * Get the static ArchiveWALRequest instance
+   * @return a ArchiveWALRequest
+   */
+  public static ArchiveWALRequest buildArchiveWALRequest() {
+    return ARCHIVE_WAL_REQUEST;
+  }
+
   /**
    * @see #buildGetServerInfoRequest()
    */

hbase-protocol-shaded/src/main/protobuf/server/region/Admin.proto

@@ -219,6 +219,12 @@ message RollWALWriterResponse {
   repeated bytes region_to_flush = 1;
 }
 
+message ArchiveWALRequest {
+}
+
+message ArchiveWALResponse {
+}
+
 message StopServerRequest {
   required string reason = 1;
 }
@@ -351,6 +357,9 @@ service AdminService {
   rpc RollWALWriter(RollWALWriterRequest)
     returns(RollWALWriterResponse);
 
+  rpc ArchiveWAL(ArchiveWALRequest)
+  returns(ArchiveWALResponse);
+
   rpc GetServerInfo(GetServerInfoRequest)
     returns(GetServerInfoResponse);
 

hbase-server/src/main/java/org/apache/hadoop/hbase/coprocessor/RegionServerObserver.java

@@ -76,6 +76,24 @@ default void postRollWALWriterRequest(
       final ObserverContext<RegionServerCoprocessorEnvironment> ctx)
       throws IOException {}
 
+  /**
+   * This will be called before executing user request to archive earliest log
+   * file of a region server.
+   * @param ctx the environment to interact with the framework and region server.
+   */
+  default void preArchiveWALRequest(
+    final ObserverContext<RegionServerCoprocessorEnvironment> ctx)
+    throws IOException {}
+
+  /**
+   * This will be called after executing user request to archive earliest log
+   * file of a region server.
+   * @param ctx the environment to interact with the framework and region server.
+   */
+  default void postArchiveWALRequest(
+    final ObserverContext<RegionServerCoprocessorEnvironment> ctx)
+    throws IOException {}
+
   /**
    * This will be called after the replication endpoint is instantiated.
    * @param ctx the environment to interact with the framework and region server.

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

@@ -171,6 +171,8 @@
 import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
 import org.apache.hadoop.hbase.shaded.protobuf.ResponseConverter;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ArchiveWALRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ArchiveWALResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearRegionBlockCacheRequest;
@@ -2394,6 +2396,27 @@ public RollWALWriterResponse rollWALWriter(final RpcController controller,
     }
   }
 
+  /**
+   * Archive the earliest log file of the region server.
+   * @param controller the RPC controller
+   * @param request the request
+   * @throws ServiceException
+   */
+  @Override
+  public ArchiveWALResponse archiveWAL(final RpcController controller,
+    final ArchiveWALRequest request) throws ServiceException {
+    try {
+      checkOpen();
+      requestCount.increment();
+      regionServer.getRegionServerCoprocessorHost().preArchiveWALRequest();
+      regionServer.getWalRoller().requestArchive();
+      regionServer.getRegionServerCoprocessorHost().postArchiveWALRequest();
+      ArchiveWALResponse.Builder builder = ArchiveWALResponse.newBuilder();
+      return builder.build();
+    } catch (IOException ie) {
+      throw new ServiceException(ie);
+    }
+  }
 
   /**
    * Stop the region server.

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

@@ -150,6 +150,24 @@ public void call(RegionServerObserver observer) throws IOException {
     });
   }
 
+  public void preArchiveWALRequest() throws IOException {
+    execOperation(coprocEnvironments.isEmpty() ? null : new RegionServerObserverOperation() {
+      @Override
+      public void call(RegionServerObserver observer) throws IOException {
+        observer.preArchiveWALRequest(this);
+      }
+    });
+  }
+
+  public void postArchiveWALRequest() throws IOException {
+    execOperation(coprocEnvironments.isEmpty() ? null : new RegionServerObserverOperation() {
+      @Override
+      public void call(RegionServerObserver observer) throws IOException {
+        observer.postArchiveWALRequest(this);
+      }
+    });
+  }
+
   public void preReplicateLogEntries()
       throws IOException {
     execOperation(coprocEnvironments.isEmpty() ? null : new RegionServerObserverOperation() {

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

@@ -57,14 +57,17 @@
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.PrivateCellUtil;
+import org.apache.hadoop.hbase.RegionException;
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.exceptions.TimeoutIOException;
 import org.apache.hadoop.hbase.io.util.MemorySizeUtil;
 import org.apache.hadoop.hbase.ipc.RpcServer;
 import org.apache.hadoop.hbase.ipc.ServerCall;
 import org.apache.hadoop.hbase.log.HBaseMarkers;
+import org.apache.hadoop.hbase.regionserver.FlushLifeCycleTracker;
 import org.apache.hadoop.hbase.regionserver.HRegion;
 import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl;
+import org.apache.hadoop.hbase.regionserver.RegionServerServices;
 import org.apache.hadoop.hbase.trace.TraceUtil;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
@@ -718,7 +721,7 @@ private void cleanOldLogs() throws IOException {
     if (logsToArchive != null) {
       for (Pair<Path, Long> logAndSize : logsToArchive) {
         this.totalLogSize.addAndGet(-logAndSize.getSecond());
-        archiveLogFile(logAndSize.getFirst());
+        moveLogFileToArchiveDir(logAndSize.getFirst());
         this.walFile2Props.remove(logAndSize.getFirst());
       }
     }
@@ -732,7 +735,7 @@ public static Path getWALArchivePath(Path archiveDir, Path p) {
     return new Path(archiveDir, p.getName());
   }
 
-  private void archiveLogFile(final Path p) throws IOException {
+  private void moveLogFileToArchiveDir(final Path p) throws IOException {
     Path newPath = getWALArchivePath(this.walArchiveDir, p);
     // Tell our listeners that a log is going to be archived.
     if (!this.listeners.isEmpty()) {
@@ -872,6 +875,33 @@ public Map<byte[], List<byte[]>> rollWriter(boolean force) throws IOException {
     }
   }
 
+  public void archive(RegionServerServices services) throws IOException{
+    if (getNumRolledLogFiles() < 1) {
+      return;
+    }
+    // get the earliest log of this WAL instance
+    Map.Entry<Path, WalProps> firstWALEntry = this.walFile2Props.firstEntry();
+    // flush regions if necessary
+    Map<byte[], List<byte[]>> regions =
+      this.sequenceIdAccounting.findLower(firstWALEntry.getValue().encodedName2HighestSequenceId);
+    if (regions != null) {
+      for (Map.Entry<byte[], List<byte[]>> entry : regions.entrySet()) {
+        String encodedRegionName = Bytes.toString(entry.getKey());
+        HRegion r = (HRegion) services.getRegion(encodedRegionName);
+        if (r == null) {
+          throw new RegionException("Failed to flush of " + encodedRegionName +
+            " when archive manually, because it is not online on rs");
+        }
+        r.flushcache(entry.getValue(), false, FlushLifeCycleTracker.DUMMY);
+      }
+    }
+
+    // move the log file to archive dir
+    moveLogFileToArchiveDir(firstWALEntry.getKey());
+    this.totalLogSize.addAndGet(-firstWALEntry.getValue().logSize);
+    this.walFile2Props.remove(firstWALEntry.getKey());
+  }
+
   // public only until class moves to o.a.h.h.wal
   /** @return the size of log files in use */
   public long getLogFileSize() {

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

@@ -30,6 +30,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.Abortable;
 import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.regionserver.RegionServerServices;
 import org.apache.hadoop.hbase.regionserver.wal.AbstractFSWAL;
 import org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException;
 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
@@ -100,6 +101,14 @@ public void requestRollAll() {
     }
   }
 
+  public void requestArchive() throws IOException {
+    synchronized (this) {
+      for (WAL wal : wals.keySet()) {
+        wal.archive((RegionServerServices) this.abortable);
+      }
+    }
+  }
+
   protected AbstractWALRoller(String name, Configuration conf, T abortable) {
     super(name);
     this.abortable = abortable;

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

@@ -32,6 +32,7 @@
 import org.apache.hadoop.hbase.PrivateCellUtil;
 import org.apache.hadoop.hbase.client.RegionInfo;
 import org.apache.hadoop.hbase.regionserver.MultiVersionConcurrencyControl.WriteEntry;
+import org.apache.hadoop.hbase.regionserver.RegionServerServices;
 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
 import org.apache.hadoop.hbase.regionserver.wal.WALCoprocessorHost;
 import org.apache.hadoop.hbase.util.CommonFSUtils;
@@ -143,6 +144,10 @@ public Map<byte[], List<byte[]>> rollWriter(boolean force) {
       return rollWriter();
     }
 
+    @Override
+    public void archive(RegionServerServices services) {
+    }
+
     @Override
     public void shutdown() {
       if(closed.compareAndSet(false, true)) {

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

@@ -24,6 +24,7 @@
 import java.util.Set;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.client.RegionInfo;
+import org.apache.hadoop.hbase.regionserver.RegionServerServices;
 import org.apache.hadoop.hbase.regionserver.wal.FailedLogCloseException;
 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
 import org.apache.hadoop.hbase.regionserver.wal.WALCoprocessorHost;
@@ -83,6 +84,11 @@ public interface WAL extends Closeable, WALFileLengthProvider {
    */
   Map<byte[], List<byte[]>> rollWriter(boolean force) throws IOException;
 
+  /**
+   * Archive the earliest log file.
+   */
+  void archive(RegionServerServices services) throws IOException;
+
   /**
    * Stop accepting new writes. If we have unsynced writes still in buffer, sync them.
    * Extant edits are left in place in backing storage to be replayed later.