You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by zg...@apache.org on 2019/05/16 01:11:00 UTC
[hbase] branch master updated: HBASE-22404 Open/Close region
request may be executed twice when master restart
This is an automated email from the ASF dual-hosted git repository.
zghao pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hbase.git
The following commit(s) were added to refs/heads/master by this push:
new bdd2fc6 HBASE-22404 Open/Close region request may be executed twice when master restart
bdd2fc6 is described below
commit bdd2fc6149c1732d4f526dd1a31749e3ab896fc4
Author: Guanghao <zg...@apache.org>
AuthorDate: Thu May 16 09:10:55 2019 +0800
HBASE-22404 Open/Close region request may be executed twice when master restart
---
.../procedure2/RemoteProcedureDispatcher.java | 11 ++--
.../hbase/master/assignment/AssignProcedure.java | 6 ++-
.../master/assignment/CloseRegionProcedure.java | 2 +-
.../master/assignment/OpenRegionProcedure.java | 2 +-
.../assignment/RegionRemoteProcedureBase.java | 16 ++++++
.../assignment/RegionTransitionProcedure.java | 4 +-
.../hbase/master/assignment/UnassignProcedure.java | 6 ++-
.../master/procedure/SplitWALRemoteProcedure.java | 10 ++--
.../SwitchRpcThrottleRemoteProcedure.java | 15 +++---
.../master/replication/RefreshPeerProcedure.java | 8 +--
.../SyncReplicationReplayWALRemoteProcedure.java | 10 ++--
.../hadoop/hbase/regionserver/HRegionServer.java | 60 ++++++++++++++++++++++
.../hadoop/hbase/regionserver/RSRpcServices.java | 21 +++++---
.../regionserver/handler/AssignRegionHandler.java | 18 ++++---
.../handler/UnassignRegionHandler.java | 16 +++---
.../procedure/TestServerRemoteProcedure.java | 8 +--
16 files changed, 159 insertions(+), 54 deletions(-)
diff --git a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/RemoteProcedureDispatcher.java b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/RemoteProcedureDispatcher.java
index 1b8ba0e..02f6d7d 100644
--- a/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/RemoteProcedureDispatcher.java
+++ b/hbase-procedure/src/main/java/org/apache/hadoop/hbase/procedure2/RemoteProcedureDispatcher.java
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.lang.Thread.UncaughtExceptionHandler;
import java.util.HashSet;
import java.util.List;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
@@ -232,8 +233,12 @@ public abstract class RemoteProcedureDispatcher<TEnv, TRemote extends Comparable
public interface RemoteProcedure<TEnv, TRemote> {
/**
* For building the remote operation.
+ * May be empty if no need to send remote call. Usually, this means the RemoteProcedure has been
+ * finished already. This is possible, as we may have already sent the procedure to RS but then
+ * the rpc connection is broken so the executeProcedures call fails, but the RS does receive the
+ * procedure and execute it and then report back, before we retry again.
*/
- RemoteOperation remoteCallBuild(TEnv env, TRemote remote);
+ Optional<RemoteOperation> remoteCallBuild(TEnv env, TRemote remote);
/**
* Called when the executeProcedure call is failed.
@@ -277,8 +282,8 @@ public abstract class RemoteProcedureDispatcher<TEnv, TRemote extends Comparable
final TRemote remote, final Set<RemoteProcedure> remoteProcedures) {
final ArrayListMultimap<Class<?>, RemoteOperation> requestByType = ArrayListMultimap.create();
for (RemoteProcedure proc : remoteProcedures) {
- RemoteOperation operation = proc.remoteCallBuild(env, remote);
- requestByType.put(operation.getClass(), operation);
+ Optional<RemoteOperation> operation = proc.remoteCallBuild(env, remote);
+ operation.ifPresent(op -> requestByType.put(op.getClass(), op));
}
return requestByType;
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignProcedure.java
index 35510d6..0079033 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/AssignProcedure.java
@@ -18,6 +18,8 @@
package org.apache.hadoop.hbase.master.assignment;
import java.io.IOException;
+import java.util.Optional;
+
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.exceptions.UnexpectedStateException;
import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
@@ -117,9 +119,9 @@ public class AssignProcedure extends RegionTransitionProcedure {
}
@Override
- public RemoteOperation remoteCallBuild(final MasterProcedureEnv env,
+ public Optional<RemoteOperation> remoteCallBuild(final MasterProcedureEnv env,
final ServerName serverName) {
- return null;
+ return Optional.empty();
}
@Override
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/CloseRegionProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/CloseRegionProcedure.java
index af31966..9124b85 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/CloseRegionProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/CloseRegionProcedure.java
@@ -61,7 +61,7 @@ public class CloseRegionProcedure extends RegionRemoteProcedureBase {
}
@Override
- public RemoteOperation remoteCallBuild(MasterProcedureEnv env, ServerName remote) {
+ public RemoteOperation newRemoteOperation() {
return new RegionCloseOperation(this, region, getProcId(), assignCandidate);
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/OpenRegionProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/OpenRegionProcedure.java
index 4f7fcef..4e850a7 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/OpenRegionProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/OpenRegionProcedure.java
@@ -57,7 +57,7 @@ public class OpenRegionProcedure extends RegionRemoteProcedureBase {
}
@Override
- public RemoteOperation remoteCallBuild(MasterProcedureEnv env, ServerName remote) {
+ public RemoteOperation newRemoteOperation() {
return new RegionOpenOperation(this, region, getProcId());
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionRemoteProcedureBase.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionRemoteProcedureBase.java
index fa00958..a3c3382 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionRemoteProcedureBase.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionRemoteProcedureBase.java
@@ -18,6 +18,8 @@
package org.apache.hadoop.hbase.master.assignment;
import java.io.IOException;
+import java.util.Optional;
+
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
@@ -32,6 +34,7 @@ import org.apache.hadoop.hbase.procedure2.ProcedureStateSerializer;
import org.apache.hadoop.hbase.procedure2.ProcedureSuspendedException;
import org.apache.hadoop.hbase.procedure2.ProcedureUtil;
import org.apache.hadoop.hbase.procedure2.ProcedureYieldException;
+import org.apache.hadoop.hbase.procedure2.RemoteProcedureDispatcher;
import org.apache.hadoop.hbase.procedure2.RemoteProcedureDispatcher.RemoteProcedure;
import org.apache.hadoop.hbase.procedure2.RemoteProcedureException;
import org.apache.hadoop.hbase.util.RetryCounter;
@@ -82,6 +85,19 @@ public abstract class RegionRemoteProcedureBase extends Procedure<MasterProcedur
}
@Override
+ public Optional<RemoteProcedureDispatcher.RemoteOperation> remoteCallBuild(MasterProcedureEnv env,
+ ServerName remote) {
+ // REPORT_SUCCEED means that this remote open/close request already executed in RegionServer.
+ // So return empty operation and RSProcedureDispatcher no need to send it again.
+ if (state == RegionRemoteProcedureBaseState.REGION_REMOTE_PROCEDURE_REPORT_SUCCEED) {
+ return Optional.empty();
+ }
+ return Optional.of(newRemoteOperation());
+ }
+
+ protected abstract RemoteProcedureDispatcher.RemoteOperation newRemoteOperation();
+
+ @Override
public void remoteOperationCompleted(MasterProcedureEnv env) {
// should not be called since we use reportRegionStateTransition to report the result
throw new UnsupportedOperationException();
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionTransitionProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionTransitionProcedure.java
index d2f8e3c..b94b45d 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionTransitionProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/RegionTransitionProcedure.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hbase.master.assignment;
import java.io.IOException;
+import java.util.Optional;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.TableName;
@@ -124,7 +125,8 @@ public abstract class RegionTransitionProcedure extends Procedure<MasterProcedur
TransitionCode code, long seqId) throws UnexpectedStateException;
@Override
- public abstract RemoteOperation remoteCallBuild(MasterProcedureEnv env, ServerName serverName);
+ public abstract Optional<RemoteOperation> remoteCallBuild(MasterProcedureEnv env,
+ ServerName serverName);
protected abstract boolean remoteCallFailed(MasterProcedureEnv env, RegionStateNode regionNode,
IOException exception);
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/UnassignProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/UnassignProcedure.java
index 6f5c4af..5b9eadd 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/UnassignProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/assignment/UnassignProcedure.java
@@ -20,6 +20,8 @@
package org.apache.hadoop.hbase.master.assignment;
import java.io.IOException;
+import java.util.Optional;
+
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.exceptions.UnexpectedStateException;
import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
@@ -126,9 +128,9 @@ public class UnassignProcedure extends RegionTransitionProcedure {
}
@Override
- public RemoteOperation remoteCallBuild(final MasterProcedureEnv env,
+ public Optional<RemoteOperation> remoteCallBuild(final MasterProcedureEnv env,
final ServerName serverName) {
- return null;
+ return Optional.empty();
}
@Override
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/SplitWALRemoteProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/SplitWALRemoteProcedure.java
index d227022..c0bd92a 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/SplitWALRemoteProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/SplitWALRemoteProcedure.java
@@ -19,6 +19,7 @@
package org.apache.hadoop.hbase.master.procedure;
import java.io.IOException;
+import java.util.Optional;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.DoNotRetryIOException;
@@ -84,11 +85,12 @@ public class SplitWALRemoteProcedure extends ServerRemoteProcedure
}
@Override
- public RemoteProcedureDispatcher.RemoteOperation remoteCallBuild(MasterProcedureEnv env,
+ public Optional<RemoteProcedureDispatcher.RemoteOperation> remoteCallBuild(MasterProcedureEnv env,
ServerName serverName) {
- return 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()));
}
@Override
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/SwitchRpcThrottleRemoteProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/SwitchRpcThrottleRemoteProcedure.java
index c69faf6..8ce9aef 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/SwitchRpcThrottleRemoteProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/procedure/SwitchRpcThrottleRemoteProcedure.java
@@ -18,6 +18,7 @@
package org.apache.hadoop.hbase.master.procedure;
import java.io.IOException;
+import java.util.Optional;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.procedure2.ProcedureStateSerializer;
@@ -74,15 +75,13 @@ public class SwitchRpcThrottleRemoteProcedure extends ServerRemoteProcedure
}
@Override
- public RemoteProcedureDispatcher.RemoteOperation
- remoteCallBuild(MasterProcedureEnv masterProcedureEnv, ServerName remote) {
+ public Optional<RemoteProcedureDispatcher.RemoteOperation> remoteCallBuild(
+ MasterProcedureEnv masterProcedureEnv, ServerName remote) {
assert targetServer.equals(remote);
- return new RSProcedureDispatcher.ServerOperation(this, getProcId(),
- SwitchRpcThrottleRemoteCallable.class,
- SwitchRpcThrottleRemoteStateData.newBuilder()
- .setTargetServer(ProtobufUtil.toServerName(remote))
- .setRpcThrottleEnabled(rpcThrottleEnabled).build()
- .toByteArray());
+ return Optional.of(new RSProcedureDispatcher.ServerOperation(this, getProcId(),
+ SwitchRpcThrottleRemoteCallable.class, SwitchRpcThrottleRemoteStateData.newBuilder()
+ .setTargetServer(ProtobufUtil.toServerName(remote))
+ .setRpcThrottleEnabled(rpcThrottleEnabled).build().toByteArray()));
}
@Override
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/RefreshPeerProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/RefreshPeerProcedure.java
index d39235f..5afe085 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/RefreshPeerProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/RefreshPeerProcedure.java
@@ -18,6 +18,8 @@
package org.apache.hadoop.hbase.master.replication;
import java.io.IOException;
+import java.util.Optional;
+
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
import org.apache.hadoop.hbase.master.procedure.PeerProcedureInterface;
@@ -112,12 +114,12 @@ public class RefreshPeerProcedure extends ServerRemoteProcedure
}
@Override
- public RemoteOperation remoteCallBuild(MasterProcedureEnv env, ServerName remote) {
+ public Optional<RemoteOperation> remoteCallBuild(MasterProcedureEnv env, ServerName remote) {
assert targetServer.equals(remote);
- return new ServerOperation(this, getProcId(), RefreshPeerCallable.class,
+ return Optional.of(new ServerOperation(this, getProcId(), RefreshPeerCallable.class,
RefreshPeerParameter.newBuilder().setPeerId(peerId).setType(toPeerModificationType(type))
.setTargetServer(ProtobufUtil.toServerName(remote)).setStage(stage).build()
- .toByteArray());
+ .toByteArray()));
}
@Override
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/SyncReplicationReplayWALRemoteProcedure.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/SyncReplicationReplayWALRemoteProcedure.java
index 7452378..6649a7e 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/SyncReplicationReplayWALRemoteProcedure.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/SyncReplicationReplayWALRemoteProcedure.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.hbase.master.replication;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
+import java.util.Optional;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
@@ -62,13 +63,14 @@ public class SyncReplicationReplayWALRemoteProcedure extends ServerRemoteProcedu
}
@Override
- public RemoteOperation remoteCallBuild(MasterProcedureEnv env, ServerName remote) {
+ public Optional<RemoteOperation> remoteCallBuild(MasterProcedureEnv env, ServerName remote) {
ReplaySyncReplicationWALParameter.Builder builder =
- ReplaySyncReplicationWALParameter.newBuilder();
+ ReplaySyncReplicationWALParameter.newBuilder();
builder.setPeerId(peerId);
wals.stream().forEach(builder::addWal);
- return new ServerOperation(this, getProcId(), ReplaySyncReplicationWALCallable.class,
- builder.build().toByteArray());
+ return Optional
+ .of(new ServerOperation(this, getProcId(), ReplaySyncReplicationWALCallable.class,
+ builder.build().toByteArray()));
}
protected void complete(MasterProcedureEnv env, Throwable error) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
index 2b1ea50..6dc5ada 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HRegionServer.java
@@ -51,6 +51,7 @@ import java.util.TreeSet;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.function.Function;
@@ -191,6 +192,8 @@ import sun.misc.SignalHandler;
import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
import org.apache.hbase.thirdparty.com.google.common.base.Preconditions;
import org.apache.hbase.thirdparty.com.google.common.base.Throwables;
+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.collect.Maps;
import org.apache.hbase.thirdparty.com.google.protobuf.BlockingRpcChannel;
import org.apache.hbase.thirdparty.com.google.protobuf.RpcController;
@@ -255,6 +258,18 @@ public class HRegionServer extends HasThread implements
protected final ConcurrentMap<byte[], Boolean> regionsInTransitionInRS =
new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR);
+ /**
+ * Used to cache the open/close region procedures which already submitted.
+ * See {@link #submitRegionProcedure(long)}.
+ */
+ private final ConcurrentMap<Long, Long> submittedRegionProcedures = new ConcurrentHashMap<>();
+ /**
+ * Used to cache the open/close region procedures which already executed.
+ * See {@link #submitRegionProcedure(long)}.
+ */
+ private final Cache<Long, Long> executedRegionProcedures =
+ CacheBuilder.newBuilder().expireAfterAccess(600, TimeUnit.SECONDS).build();
+
// Cache flushing
protected MemStoreFlusher cacheFlusher;
@@ -3882,6 +3897,51 @@ public class HRegionServer extends HasThread implements
}
}
+ /**
+ * Will ignore the open/close region procedures which already submitted or executed.
+ *
+ * When master had unfinished open/close region procedure and restarted, new active master may
+ * send duplicate open/close region request to regionserver. The open/close request is submitted
+ * to a thread pool and execute. So first need a cache for submitted open/close region procedures.
+ *
+ * After the open/close region request executed and report region transition succeed, cache it in
+ * executed region procedures cache. See {@link #finishRegionProcedure(long)}. After report region
+ * transition succeed, master will not send the open/close region request to regionserver again.
+ * And we thought that the ongoing duplicate open/close region request should not be delayed more
+ * than 600 seconds. So the executed region procedures cache will expire after 600 seconds.
+ *
+ * See HBASE-22404 for more details.
+ *
+ * @param procId the id of the open/close region procedure
+ * @return true if the procedure can be submitted.
+ */
+ boolean submitRegionProcedure(long procId) {
+ if (procId == -1) {
+ return true;
+ }
+ // Ignore the region procedures which already submitted.
+ Long previous = submittedRegionProcedures.putIfAbsent(procId, procId);
+ if (previous != null) {
+ LOG.warn("Received procedure pid={}, which already submitted, just ignore it", procId);
+ return false;
+ }
+ // Ignore the region procedures which already executed.
+ if (executedRegionProcedures.getIfPresent(procId) != null) {
+ LOG.warn("Received procedure pid={}, which already executed, just ignore it", procId);
+ return false;
+ }
+ return true;
+ }
+
+ /**
+ * See {@link #submitRegionProcedure(long)}.
+ * @param procId the id of the open/close region procedure
+ */
+ public void finishRegionProcedure(long procId) {
+ executedRegionProcedures.put(procId, procId);
+ submittedRegionProcedures.remove(procId);
+ }
+
public boolean isShutDown() {
return shutDown;
}
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
index 688c03d..35fc7e4 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/RSRpcServices.java
@@ -3721,8 +3721,12 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
regionServer.updateRegionFavoredNodesMapping(regionInfo.getEncodedName(),
regionOpenInfo.getFavoredNodesList());
}
- regionServer.executorService.submit(AssignRegionHandler.create(regionServer, regionInfo,
- regionOpenInfo.getOpenProcId(), tableDesc, masterSystemTime));
+ long procId = regionOpenInfo.getOpenProcId();
+ if (regionServer.submitRegionProcedure(procId)) {
+ regionServer.executorService.submit(AssignRegionHandler
+ .create(regionServer, regionInfo, procId, tableDesc,
+ masterSystemTime));
+ }
}
}
@@ -3733,11 +3737,14 @@ public class RSRpcServices implements HBaseRPCErrorHandler,
} catch (DoNotRetryIOException e) {
throw new UncheckedIOException("Should not happen", e);
}
- ServerName destination =
- request.hasDestinationServer() ? ProtobufUtil.toServerName(request.getDestinationServer())
- : null;
- regionServer.executorService.submit(UnassignRegionHandler.create(regionServer, encodedName,
- request.getCloseProcId(), false, destination));
+ ServerName destination = request.hasDestinationServer() ?
+ ProtobufUtil.toServerName(request.getDestinationServer()) :
+ null;
+ long procId = request.getCloseProcId();
+ if (regionServer.submitRegionProcedure(procId)) {
+ regionServer.executorService.submit(UnassignRegionHandler
+ .create(regionServer, encodedName, procId, false, destination));
+ }
}
private void executeProcedures(RemoteProcedureRequest request) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/AssignRegionHandler.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/AssignRegionHandler.java
index a978d72..76e8f80 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/AssignRegionHandler.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/AssignRegionHandler.java
@@ -26,8 +26,8 @@ import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.executor.EventHandler;
import org.apache.hadoop.hbase.executor.EventType;
import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.hadoop.hbase.regionserver.Region;
-import org.apache.hadoop.hbase.regionserver.RegionServerServices;
import org.apache.hadoop.hbase.regionserver.RegionServerServices.PostOpenDeployContext;
import org.apache.hadoop.hbase.regionserver.RegionServerServices.RegionStateTransitionContext;
import org.apache.hadoop.hbase.util.RetryCounter;
@@ -59,7 +59,7 @@ public class AssignRegionHandler extends EventHandler {
private final RetryCounter retryCounter;
- public AssignRegionHandler(RegionServerServices server, RegionInfo regionInfo, long openProcId,
+ public AssignRegionHandler(HRegionServer server, RegionInfo regionInfo, long openProcId,
@Nullable TableDescriptor tableDesc, long masterSystemTime, EventType eventType) {
super(server, eventType);
this.regionInfo = regionInfo;
@@ -69,14 +69,14 @@ public class AssignRegionHandler extends EventHandler {
this.retryCounter = HandlerUtil.getRetryCounter();
}
- private RegionServerServices getServer() {
- return (RegionServerServices) server;
+ private HRegionServer getServer() {
+ return (HRegionServer) server;
}
private void cleanUpAndReportFailure(IOException error) throws IOException {
LOG.warn("Failed to open region {}, will report to master", regionInfo.getRegionNameAsString(),
error);
- RegionServerServices rs = getServer();
+ HRegionServer rs = getServer();
rs.getRegionsInTransitionInRS().remove(regionInfo.getEncodedNameAsBytes(), Boolean.TRUE);
if (!rs.reportRegionStateTransition(new RegionStateTransitionContext(TransitionCode.FAILED_OPEN,
HConstants.NO_SEQNUM, openProcId, masterSystemTime, regionInfo))) {
@@ -87,7 +87,7 @@ public class AssignRegionHandler extends EventHandler {
@Override
public void process() throws IOException {
- RegionServerServices rs = getServer();
+ HRegionServer rs = getServer();
String encodedName = regionInfo.getEncodedName();
byte[] encodedNameBytes = regionInfo.getEncodedNameAsBytes();
String regionName = regionInfo.getRegionNameAsString();
@@ -101,7 +101,6 @@ public class AssignRegionHandler extends EventHandler {
// reportRegionStateTransition any more.
return;
}
- LOG.info("Open {}", regionName);
Boolean previous = rs.getRegionsInTransitionInRS().putIfAbsent(encodedNameBytes, Boolean.TRUE);
if (previous != null) {
if (previous) {
@@ -121,6 +120,7 @@ public class AssignRegionHandler extends EventHandler {
}
return;
}
+ LOG.info("Open {}", regionName);
HRegion region;
try {
TableDescriptor htd =
@@ -139,6 +139,8 @@ public class AssignRegionHandler extends EventHandler {
rs.postOpenDeployTasks(new PostOpenDeployContext(region, openProcId, masterSystemTime));
rs.addRegion(region);
LOG.info("Opened {}", regionName);
+ // Cache the open region procedure id after report region transition succeed.
+ rs.finishRegionProcedure(openProcId);
Boolean current = rs.getRegionsInTransitionInRS().remove(regionInfo.getEncodedNameAsBytes());
if (current == null) {
// Should NEVER happen, but let's be paranoid.
@@ -158,7 +160,7 @@ public class AssignRegionHandler extends EventHandler {
"Failed to open region " + regionInfo.getRegionNameAsString() + " and can not recover", t);
}
- public static AssignRegionHandler create(RegionServerServices server, RegionInfo regionInfo,
+ public static AssignRegionHandler create(HRegionServer server, RegionInfo regionInfo,
long openProcId, TableDescriptor tableDesc, long masterSystemTime) {
EventType eventType;
if (regionInfo.isMetaRegion()) {
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/UnassignRegionHandler.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/UnassignRegionHandler.java
index 1d470c3..0fc8edc 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/UnassignRegionHandler.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/handler/UnassignRegionHandler.java
@@ -25,8 +25,8 @@ import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.executor.EventHandler;
import org.apache.hadoop.hbase.executor.EventType;
import org.apache.hadoop.hbase.regionserver.HRegion;
+import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.hadoop.hbase.regionserver.Region;
-import org.apache.hadoop.hbase.regionserver.RegionServerServices;
import org.apache.hadoop.hbase.regionserver.RegionServerServices.RegionStateTransitionContext;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.RetryCounter;
@@ -60,7 +60,7 @@ public class UnassignRegionHandler extends EventHandler {
private final RetryCounter retryCounter;
- public UnassignRegionHandler(RegionServerServices server, String encodedName, long closeProcId,
+ public UnassignRegionHandler(HRegionServer server, String encodedName, long closeProcId,
boolean abort, @Nullable ServerName destination, EventType eventType) {
super(server, eventType);
this.encodedName = encodedName;
@@ -70,13 +70,13 @@ public class UnassignRegionHandler extends EventHandler {
this.retryCounter = HandlerUtil.getRetryCounter();
}
- private RegionServerServices getServer() {
- return (RegionServerServices) server;
+ private HRegionServer getServer() {
+ return (HRegionServer) server;
}
@Override
public void process() throws IOException {
- RegionServerServices rs = getServer();
+ HRegionServer rs = getServer();
byte[] encodedNameBytes = Bytes.toBytes(encodedName);
Boolean previous = rs.getRegionsInTransitionInRS().putIfAbsent(encodedNameBytes, Boolean.FALSE);
if (previous != null) {
@@ -94,7 +94,7 @@ public class UnassignRegionHandler extends EventHandler {
}
return;
}
- HRegion region = (HRegion) rs.getRegion(encodedName);
+ HRegion region = rs.getRegion(encodedName);
if (region == null) {
LOG.debug(
"Received CLOSE for a region {} which is not online, and we're not opening/closing.",
@@ -124,6 +124,8 @@ public class UnassignRegionHandler extends EventHandler {
HConstants.NO_SEQNUM, closeProcId, -1, region.getRegionInfo()))) {
throw new IOException("Failed to report close to master: " + regionName);
}
+ // Cache the close region procedure id after report region transition succeed.
+ rs.finishRegionProcedure(closeProcId);
rs.getRegionsInTransitionInRS().remove(encodedNameBytes, Boolean.FALSE);
LOG.info("Closed {}", regionName);
}
@@ -134,7 +136,7 @@ public class UnassignRegionHandler extends EventHandler {
getServer().abort("Failed to close region " + encodedName + " and can not recover", t);
}
- public static UnassignRegionHandler create(RegionServerServices server, String encodedName,
+ public static UnassignRegionHandler create(HRegionServer server, String encodedName,
long closeProcId, boolean abort, @Nullable ServerName destination) {
// Just try our best to determine whether it is for closing meta. It is not the end of the world
// if we put the handler into a wrong executor.
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestServerRemoteProcedure.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestServerRemoteProcedure.java
index f03794a..14342d4 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestServerRemoteProcedure.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/master/procedure/TestServerRemoteProcedure.java
@@ -23,6 +23,7 @@ import static org.junit.Assert.fail;
import java.io.IOException;
import java.util.NavigableMap;
+import java.util.Optional;
import java.util.Set;
import java.util.SortedSet;
import java.util.concurrent.ConcurrentSkipListMap;
@@ -189,9 +190,10 @@ public class TestServerRemoteProcedure {
}
@Override
- public RemoteProcedureDispatcher.RemoteOperation remoteCallBuild(MasterProcedureEnv env,
- ServerName serverName) {
- return new RSProcedureDispatcher.ServerOperation(null, 0L, this.getClass(), new byte[0]);
+ public Optional<RemoteProcedureDispatcher.RemoteOperation> remoteCallBuild(
+ MasterProcedureEnv env, ServerName serverName) {
+ return Optional
+ .of(new RSProcedureDispatcher.ServerOperation(null, 0L, this.getClass(), new byte[0]));
}
@Override