HBASE-28690 add initiatingMasterActiveTime as a member of RemoteOperation

Author: ukumawat

hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/RemoteProcedureDispatcher.java

@@ -222,14 +222,22 @@ protected final void submitTask(Runnable task, long delay, TimeUnit unit) {
    */
   public static abstract class RemoteOperation {
     private final RemoteProcedure remoteProcedure;
+    // active time of the master that sent this request, used for fencing
+    private final long initiatingMasterActiveTime;
 
-    protected RemoteOperation(final RemoteProcedure remoteProcedure) {
+    protected RemoteOperation(final RemoteProcedure remoteProcedure,
+      long initiatingMasterActiveTime) {
       this.remoteProcedure = remoteProcedure;
+      this.initiatingMasterActiveTime = initiatingMasterActiveTime;
     }
 
     public RemoteProcedure getRemoteProcedure() {
       return remoteProcedure;
     }
+
+    public long getInitiatingMasterActiveTime() {
+      return initiatingMasterActiveTime;
+    }
   }
 
   /**

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/CloseRegionProcedure.java

@@ -64,8 +64,9 @@ public TableOperationType getTableOperationType() {
   }
 
   @Override
-  public RemoteOperation newRemoteOperation() {
-    return new RegionCloseOperation(this, region, getProcId(), assignCandidate, evictCache);
+  public RemoteOperation newRemoteOperation(MasterProcedureEnv env) {
+    return new RegionCloseOperation(this, region, getProcId(), assignCandidate, evictCache,
+      env.getMasterServices().getMasterActiveTime());
   }
 
   @Override

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/OpenRegionProcedure.java

@@ -57,8 +57,9 @@ public TableOperationType getTableOperationType() {
   }
 
   @Override
-  public RemoteOperation newRemoteOperation() {
-    return new RegionOpenOperation(this, region, getProcId());
+  public RemoteOperation newRemoteOperation(MasterProcedureEnv env) {
+    return new RegionOpenOperation(this, region, getProcId(),
+      env.getMasterServices().getMasterActiveTime());
   }
 
   @Override

hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionRemoteProcedureBase.java

@@ -92,10 +92,11 @@ public Optional<RemoteProcedureDispatcher.RemoteOperation> remoteCallBuild(Maste
     if (state == RegionRemoteProcedureBaseState.REGION_REMOTE_PROCEDURE_REPORT_SUCCEED) {
       return Optional.empty();
     }
-    return Optional.of(newRemoteOperation());
+    return Optional.of(newRemoteOperation(env));
   }
 
-  protected abstract RemoteProcedureDispatcher.RemoteOperation newRemoteOperation();
+  protected abstract RemoteProcedureDispatcher.RemoteOperation
+    newRemoteOperation(MasterProcedureEnv env);
 
   @Override
   public void remoteOperationCompleted(MasterProcedureEnv env) {

hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/FlushRegionProcedure.java

@@ -222,8 +222,9 @@ public Optional<RemoteOperation> remoteCallBuild(MasterProcedureEnv env, ServerN
         }
       }
     }
-    return Optional.of(new RSProcedureDispatcher.ServerOperation(this, getProcId(),
-      FlushRegionCallable.class, builder.build().toByteArray()));
+    return Optional
+      .of(new RSProcedureDispatcher.ServerOperation(this, getProcId(), FlushRegionCallable.class,
+        builder.build().toByteArray(), env.getMasterServices().getMasterActiveTime()));
   }
 
   @Override

hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/RSProcedureDispatcher.java

@@ -421,15 +421,13 @@ public void dispatchOpenRequests(final MasterProcedureEnv env,
     public void dispatchCloseRequests(final MasterProcedureEnv env,
       final List<RegionCloseOperation> operations) {
       for (RegionCloseOperation op : operations) {
-        request.addCloseRegion(op.buildCloseRegionRequest(getServerName(),
-          env.getMasterServices().getMasterActiveTime()));
+        request.addCloseRegion(op.buildCloseRegionRequest(getServerName()));
       }
     }
 
     @Override
     public void dispatchServerOperations(MasterProcedureEnv env, List<ServerOperation> operations) {
-      operations.stream().map(o -> o.buildRequest(env.getMasterServices().getMasterActiveTime()))
-        .forEachOrdered(request::addProc);
+      operations.stream().map(ServerOperation::buildRequest).forEachOrdered(request::addProc);
     }
 
     // will be overridden in test.
@@ -453,7 +451,8 @@ private static OpenRegionRequest buildOpenRegionRequest(final MasterProcedureEnv
     final ServerName serverName, final List<RegionOpenOperation> operations) {
     final OpenRegionRequest.Builder builder = OpenRegionRequest.newBuilder();
     builder.setServerStartCode(serverName.getStartCode());
-    builder.setInitiatingMasterActiveTime(env.getMasterServices().getMasterActiveTime());
+    builder.setInitiatingMasterActiveTime(operations.stream()
+      .map(RemoteOperation::getInitiatingMasterActiveTime).findAny().orElse(-1L));
     builder.setMasterSystemTime(EnvironmentEdgeManager.currentTime());
     for (RegionOpenOperation op : operations) {
       builder.addOpenInfo(op.buildRegionOpenInfoRequest(env));
@@ -476,36 +475,37 @@ public static final class ServerOperation extends RemoteOperation {
     private final byte[] rsProcData;
 
     public ServerOperation(RemoteProcedure remoteProcedure, long procId, Class<?> rsProcClass,
-      byte[] rsProcData) {
-      super(remoteProcedure);
+      byte[] rsProcData, long initiatingMasterActiveTime) {
+      super(remoteProcedure, initiatingMasterActiveTime);
       this.procId = procId;
       this.rsProcClass = rsProcClass;
       this.rsProcData = rsProcData;
     }
 
-    public RemoteProcedureRequest buildRequest(long initiatingMasterActiveTime) {
+    public RemoteProcedureRequest buildRequest() {
       return RemoteProcedureRequest.newBuilder().setProcId(procId)
         .setProcClass(rsProcClass.getName()).setProcData(ByteString.copyFrom(rsProcData))
-        .setInitiatingMasterActiveTime(initiatingMasterActiveTime).build();
+        .setInitiatingMasterActiveTime(getInitiatingMasterActiveTime()).build();
     }
   }
 
   public static abstract class RegionOperation extends RemoteOperation {
     protected final RegionInfo regionInfo;
     protected final long procId;
 
-    protected RegionOperation(RemoteProcedure remoteProcedure, RegionInfo regionInfo, long procId) {
-      super(remoteProcedure);
+    protected RegionOperation(RemoteProcedure remoteProcedure, RegionInfo regionInfo, long procId,
+      long initiatingMasterActiveTime) {
+      super(remoteProcedure, initiatingMasterActiveTime);
       this.regionInfo = regionInfo;
       this.procId = procId;
     }
   }
 
   public static class RegionOpenOperation extends RegionOperation {
 
-    public RegionOpenOperation(RemoteProcedure remoteProcedure, RegionInfo regionInfo,
-      long procId) {
-      super(remoteProcedure, regionInfo, procId);
+    public RegionOpenOperation(RemoteProcedure remoteProcedure, RegionInfo regionInfo, long procId,
+      long initiatingMasterActiveTime) {
+      super(remoteProcedure, regionInfo, procId, initiatingMasterActiveTime);
     }
 
     public OpenRegionRequest.RegionOpenInfo
@@ -520,8 +520,8 @@ public static class RegionCloseOperation extends RegionOperation {
     private boolean evictCache;
 
     public RegionCloseOperation(RemoteProcedure remoteProcedure, RegionInfo regionInfo, long procId,
-      ServerName destinationServer, boolean evictCache) {
-      super(remoteProcedure, regionInfo, procId);
+      ServerName destinationServer, boolean evictCache, long initiatingMasterActiveTime) {
+      super(remoteProcedure, regionInfo, procId, initiatingMasterActiveTime);
       this.destinationServer = destinationServer;
       this.evictCache = evictCache;
     }
@@ -530,11 +530,9 @@ public ServerName getDestinationServer() {
       return destinationServer;
     }
 
-    public CloseRegionRequest buildCloseRegionRequest(final ServerName serverName,
-      long initiatingMasterActiveTime) {
+    public CloseRegionRequest buildCloseRegionRequest(final ServerName serverName) {
       return ProtobufUtil.buildCloseRegionRequest(serverName, regionInfo.getRegionName(),
-        getDestinationServer(), procId, evictCache, initiatingMasterActiveTime);
-
+        getDestinationServer(), procId, evictCache, getInitiatingMasterActiveTime());
     }
   }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/SnapshotRegionProcedure.java

@@ -95,9 +95,11 @@ protected boolean holdLock(MasterProcedureEnv env) {
 
   @Override
   public Optional<RemoteOperation> remoteCallBuild(MasterProcedureEnv env, ServerName serverName) {
-    return Optional.of(new RSProcedureDispatcher.ServerOperation(this, getProcId(),
-      SnapshotRegionCallable.class, MasterProcedureProtos.SnapshotRegionParameter.newBuilder()
-        .setRegion(ProtobufUtil.toRegionInfo(region)).setSnapshot(snapshot).build().toByteArray()));
+    return Optional
+      .of(new RSProcedureDispatcher.ServerOperation(this, getProcId(), SnapshotRegionCallable.class,
+        MasterProcedureProtos.SnapshotRegionParameter.newBuilder()
+          .setRegion(ProtobufUtil.toRegionInfo(region)).setSnapshot(snapshot).build().toByteArray(),
+        env.getMasterServices().getMasterActiveTime()));
   }
 
   @Override

hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/SnapshotVerifyProcedure.java

@@ -224,8 +224,9 @@ protected void toStringClassDetails(StringBuilder builder) {
   public Optional<RemoteOperation> remoteCallBuild(MasterProcedureEnv env, ServerName serverName) {
     SnapshotVerifyParameter.Builder builder = SnapshotVerifyParameter.newBuilder();
     builder.setSnapshot(snapshot).setRegion(ProtobufUtil.toRegionInfo(region));
-    return Optional.of(new RSProcedureDispatcher.ServerOperation(this, getProcId(),
-      SnapshotVerifyCallable.class, builder.build().toByteArray()));
+    return Optional
+      .of(new RSProcedureDispatcher.ServerOperation(this, getProcId(), SnapshotVerifyCallable.class,
+        builder.build().toByteArray(), env.getMasterServices().getMasterActiveTime()));
   }
 
   @Override

hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/SplitWALRemoteProcedure.java

@@ -97,9 +97,10 @@ protected void deserializeStateData(ProcedureStateSerializer serializer) throws
   @Override
   public Optional<RemoteProcedureDispatcher.RemoteOperation> remoteCallBuild(MasterProcedureEnv env,
     ServerName serverName) {
-    return Optional.of(new RSProcedureDispatcher.ServerOperation(this, getProcId(),
-      SplitWALCallable.class, MasterProcedureProtos.SplitWALParameter.newBuilder()
-        .setWalPath(walPath).build().toByteArray()));
+    return Optional.of(new RSProcedureDispatcher.ServerOperation(
+      this, getProcId(), SplitWALCallable.class, MasterProcedureProtos.SplitWALParameter
+        .newBuilder().setWalPath(walPath).build().toByteArray(),
+      env.getMasterServices().getMasterActiveTime()));
   }
 
   @Override

hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/SwitchRpcThrottleRemoteProcedure.java

@@ -93,7 +93,8 @@ protected void deserializeStateData(ProcedureStateSerializer serializer) throws
       SwitchRpcThrottleRemoteCallable.class,
       SwitchRpcThrottleRemoteStateData.newBuilder()
         .setTargetServer(ProtobufUtil.toServerName(remote))
-        .setRpcThrottleEnabled(rpcThrottleEnabled).build().toByteArray()));
+        .setRpcThrottleEnabled(rpcThrottleEnabled).build().toByteArray(),
+      masterProcedureEnv.getMasterServices().getMasterActiveTime()));
   }
 
   @Override