HBASE-15843 Replace RegionState.getRegionInTransition() Map with a Set
This commit is contained in:
parent
a9972355c4
commit
e5eea96ca6
@ -24,7 +24,9 @@ import java.util.Arrays;
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.HashMap;
|
||||
import java.util.HashSet;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
import org.apache.hadoop.hbase.util.ByteStringer;
|
||||
import org.apache.hadoop.hbase.classification.InterfaceAudience;
|
||||
@ -80,7 +82,7 @@ public class ClusterStatus extends VersionedWritable {
|
||||
private Collection<ServerName> deadServers;
|
||||
private ServerName master;
|
||||
private Collection<ServerName> backupMasters;
|
||||
private Map<String, RegionState> intransition;
|
||||
private Set<RegionState> intransition;
|
||||
private String clusterId;
|
||||
private String[] masterCoprocessors;
|
||||
private Boolean balancerOn;
|
||||
@ -102,7 +104,7 @@ public class ClusterStatus extends VersionedWritable {
|
||||
final Collection<ServerName> deadServers,
|
||||
final ServerName master,
|
||||
final Collection<ServerName> backupMasters,
|
||||
final Map<String, RegionState> rit,
|
||||
final Set<RegionState> rit,
|
||||
final String[] masterCoprocessors,
|
||||
final Boolean balancerOn) {
|
||||
this.hbaseVersion = hbaseVersion;
|
||||
@ -275,7 +277,7 @@ public class ClusterStatus extends VersionedWritable {
|
||||
}
|
||||
|
||||
@InterfaceAudience.Private
|
||||
public Map<String, RegionState> getRegionsInTransition() {
|
||||
public Set<RegionState> getRegionsInTransition() {
|
||||
return this.intransition;
|
||||
}
|
||||
|
||||
@ -354,7 +356,7 @@ public class ClusterStatus extends VersionedWritable {
|
||||
int ritSize = (intransition != null) ? intransition.size() : 0;
|
||||
sb.append("\nNumber of regions in transition: " + ritSize);
|
||||
if (ritSize > 0) {
|
||||
for (RegionState state: intransition.values()) {
|
||||
for (RegionState state: intransition) {
|
||||
sb.append("\n " + state.toDescriptiveString());
|
||||
}
|
||||
}
|
||||
@ -388,11 +390,11 @@ public class ClusterStatus extends VersionedWritable {
|
||||
}
|
||||
|
||||
if (intransition != null) {
|
||||
for (Map.Entry<String, RegionState> rit : getRegionsInTransition().entrySet()) {
|
||||
ClusterStatusProtos.RegionState rs = rit.getValue().convert();
|
||||
for (RegionState rit : getRegionsInTransition()) {
|
||||
ClusterStatusProtos.RegionState rs = rit.convert();
|
||||
RegionSpecifier.Builder spec =
|
||||
RegionSpecifier.newBuilder().setType(RegionSpecifierType.REGION_NAME);
|
||||
spec.setValue(ByteStringer.wrap(Bytes.toBytes(rit.getKey())));
|
||||
spec.setValue(ByteStringer.wrap(rit.getRegion().getRegionName()));
|
||||
|
||||
RegionInTransition pbRIT =
|
||||
RegionInTransition.newBuilder().setSpec(spec.build()).setRegionState(rs).build();
|
||||
@ -461,13 +463,12 @@ public class ClusterStatus extends VersionedWritable {
|
||||
}
|
||||
}
|
||||
|
||||
Map<String, RegionState> rit = null;
|
||||
Set<RegionState> rit = null;
|
||||
if (proto.getRegionsInTransitionList() != null) {
|
||||
rit = new HashMap<String, RegionState>(proto.getRegionsInTransitionList().size());
|
||||
rit = new HashSet<RegionState>(proto.getRegionsInTransitionList().size());
|
||||
for (RegionInTransition region : proto.getRegionsInTransitionList()) {
|
||||
String key = new String(region.getSpec().getValue().toByteArray());
|
||||
RegionState value = RegionState.convert(region.getRegionState());
|
||||
rit.put(key, value);
|
||||
rit.add(value);
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -1714,7 +1714,7 @@ public final class ProtobufUtil {
|
||||
* @param transitionInZK
|
||||
* @throws IOException
|
||||
*/
|
||||
public static void closeRegion(final RpcController controller,
|
||||
public static void closeRegion(final RpcController controller,
|
||||
final AdminService.BlockingInterface admin, final ServerName server, final byte[] regionName,
|
||||
final boolean transitionInZK) throws IOException {
|
||||
CloseRegionRequest closeRegionRequest =
|
||||
@ -2947,7 +2947,7 @@ public final class ProtobufUtil {
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Convert a protocol buffer TimeUnit to a client TimeUnit
|
||||
* @param proto
|
||||
@ -3185,7 +3185,7 @@ public final class ProtobufUtil {
|
||||
* @param builder current message builder
|
||||
* @param in InputStream containing protobuf data
|
||||
* @param size known size of protobuf data
|
||||
* @throws IOException
|
||||
* @throws IOException
|
||||
*/
|
||||
public static void mergeFrom(Message.Builder builder, InputStream in, int size)
|
||||
throws IOException {
|
||||
@ -3200,7 +3200,7 @@ public final class ProtobufUtil {
|
||||
* buffers where the message size is not known
|
||||
* @param builder current message builder
|
||||
* @param in InputStream containing protobuf data
|
||||
* @throws IOException
|
||||
* @throws IOException
|
||||
*/
|
||||
public static void mergeFrom(Message.Builder builder, InputStream in)
|
||||
throws IOException {
|
||||
@ -3214,8 +3214,8 @@ public final class ProtobufUtil {
|
||||
* This version of protobuf's mergeFrom avoids the hard-coded 64MB limit for decoding
|
||||
* buffers when working with ByteStrings
|
||||
* @param builder current message builder
|
||||
* @param bs ByteString containing the
|
||||
* @throws IOException
|
||||
* @param bs ByteString containing the
|
||||
* @throws IOException
|
||||
*/
|
||||
public static void mergeFrom(Message.Builder builder, ByteString bs) throws IOException {
|
||||
final CodedInputStream codedInput = bs.newCodedInput();
|
||||
@ -3229,7 +3229,7 @@ public final class ProtobufUtil {
|
||||
* buffers when working with byte arrays
|
||||
* @param builder current message builder
|
||||
* @param b byte array
|
||||
* @throws IOException
|
||||
* @throws IOException
|
||||
*/
|
||||
public static void mergeFrom(Message.Builder builder, byte[] b) throws IOException {
|
||||
final CodedInputStream codedInput = CodedInputStream.newInstance(b);
|
||||
|
@ -24,15 +24,14 @@ org.apache.hadoop.conf.Configuration;
|
||||
org.apache.hadoop.hbase.HBaseConfiguration;
|
||||
org.apache.hadoop.hbase.HConstants;
|
||||
java.util.Iterator;
|
||||
java.util.Map;
|
||||
java.util.Set;
|
||||
</%import>
|
||||
<%args>
|
||||
AssignmentManager assignmentManager;
|
||||
int limit = 100;
|
||||
</%args>
|
||||
<%java>
|
||||
Map<String, RegionState> rit = assignmentManager
|
||||
.getRegionStates().getRegionsInTransition();
|
||||
Set<RegionState> rit = assignmentManager.getRegionStates().getRegionsInTransition();
|
||||
// process the map to find region in transition details
|
||||
Configuration conf = HBaseConfiguration.create();
|
||||
int ritThreshold = conf.getInt(HConstants.METRICS_RIT_STUCK_WARNING_THRESHOLD, 60000);
|
||||
@ -40,14 +39,14 @@ int numOfRITOverThreshold = 0;
|
||||
long maxRITTime = Long.MIN_VALUE;
|
||||
long currentTime = System.currentTimeMillis();
|
||||
String regionIDForOldestRIT = ""; // avoiding null
|
||||
for (Map.Entry<String, RegionState> e : rit.entrySet()) {
|
||||
long ritTime = currentTime - e.getValue().getStamp();
|
||||
for (RegionState rs : rit) {
|
||||
long ritTime = currentTime - rs.getStamp();
|
||||
if(ritTime > ritThreshold) {
|
||||
numOfRITOverThreshold++;
|
||||
}
|
||||
if(maxRITTime < ritTime) {
|
||||
maxRITTime = ritTime;
|
||||
regionIDForOldestRIT = e.getKey();
|
||||
regionIDForOldestRIT = rs.getRegion().getEncodedName();
|
||||
}
|
||||
}
|
||||
|
||||
@ -56,16 +55,14 @@ int toRemove = rit.size() - limit;
|
||||
int removed = 0;
|
||||
if (toRemove > 0) {
|
||||
// getRegionsInTransition returned a copy, so we can mutate it
|
||||
for (Iterator<Map.Entry<String, RegionState>> it = rit.entrySet().iterator();
|
||||
it.hasNext() && toRemove > 0;
|
||||
) {
|
||||
Map.Entry<String, RegionState> e = it.next();
|
||||
if (HRegionInfo.FIRST_META_REGIONINFO.getEncodedName().equals(
|
||||
e.getKey()) ||
|
||||
regionIDForOldestRIT.equals(e.getKey())) {
|
||||
for (Iterator<RegionState> it = rit.iterator(); it.hasNext() && toRemove > 0;) {
|
||||
RegionState rs = it.next();
|
||||
String regionEncodedName = rs.getRegion().getEncodedName();
|
||||
if (HRegionInfo.FIRST_META_REGIONINFO.getEncodedName().equals(regionEncodedName) ||
|
||||
regionIDForOldestRIT.equals(regionEncodedName)) {
|
||||
// don't remove the meta & the oldest rit regions, they're too interesting!
|
||||
continue;
|
||||
}
|
||||
}
|
||||
it.remove();
|
||||
toRemove--;
|
||||
removed++;
|
||||
@ -80,14 +77,14 @@ if (toRemove > 0) {
|
||||
<h2>Regions in Transition</h2>
|
||||
<table class="table table-striped">
|
||||
<tr><th>Region</th><th>State</th><th>RIT time (ms)</th></tr>
|
||||
<%for Map.Entry<String, RegionState> entry : rit.entrySet() %>
|
||||
<%if regionIDForOldestRIT.equals(entry.getKey()) %>
|
||||
<%for RegionState rs : rit %>
|
||||
<%if regionIDForOldestRIT.equals(rs.getRegion().getEncodedName()) %>
|
||||
<tr BGCOLOR="#FE2E2E" >
|
||||
<%else>
|
||||
<tr>
|
||||
</%if>
|
||||
<td><% entry.getKey() %></td><td><% entry.getValue().toDescriptiveString() %></td>
|
||||
<td><% (currentTime - entry.getValue().getStamp()) %> </td></tr>
|
||||
<td><% rs.getRegion().getEncodedName() %></td><td><% rs.toDescriptiveString() %></td>
|
||||
<td><% (currentTime - rs.getStamp()) %> </td></tr>
|
||||
</%for>
|
||||
<%if numOfRITOverThreshold > 0 %>
|
||||
<tr BGCOLOR="#D7DF01" >
|
||||
|
@ -558,10 +558,10 @@ public class AssignmentManager extends ZooKeeperListener {
|
||||
}
|
||||
if (!failover && !useZKForAssignment) {
|
||||
// If any region except meta is in transition on a live server, it's a failover.
|
||||
Map<String, RegionState> regionsInTransition = regionStates.getRegionsInTransition();
|
||||
Set<RegionState> regionsInTransition = regionStates.getRegionsInTransition();
|
||||
if (!regionsInTransition.isEmpty()) {
|
||||
Set<ServerName> onlineServers = serverManager.getOnlineServers().keySet();
|
||||
for (RegionState regionState: regionsInTransition.values()) {
|
||||
for (RegionState regionState: regionsInTransition) {
|
||||
ServerName serverName = regionState.getServerName();
|
||||
if (!regionState.getRegion().isMetaRegion()
|
||||
&& serverName != null && onlineServers.contains(serverName)) {
|
||||
@ -2312,8 +2312,8 @@ public class AssignmentManager extends ZooKeeperListener {
|
||||
LOG.debug("ALREADY_OPENED " + region.getRegionNameAsString()
|
||||
+ " to " + sn);
|
||||
String encodedName = region.getEncodedName();
|
||||
|
||||
//If use ZkForAssignment, region already Opened event should not be handled,
|
||||
|
||||
//If use ZkForAssignment, region already Opened event should not be handled,
|
||||
//leave it to zk event. See HBase-14407.
|
||||
if(useZKForAssignment){
|
||||
String node = ZKAssign.getNodeName(watcher, encodedName);
|
||||
@ -2335,10 +2335,10 @@ public class AssignmentManager extends ZooKeeperListener {
|
||||
} catch (DeserializationException e) {
|
||||
LOG.warn("Get RegionTransition from zk deserialization failed! ", e);
|
||||
}
|
||||
|
||||
|
||||
deleteNodeInStates(encodedName, "offline", sn, EventType.M_ZK_REGION_OFFLINE);
|
||||
}
|
||||
|
||||
|
||||
regionStates.regionOnline(region, sn);
|
||||
}
|
||||
|
||||
@ -3142,8 +3142,8 @@ public class AssignmentManager extends ZooKeeperListener {
|
||||
// since we update the state before we send the RPC call. We can't update
|
||||
// the state after the RPC call. Otherwise, we don't know what's happened
|
||||
// to the region if the master dies right after the RPC call is out.
|
||||
Map<String, RegionState> rits = regionStates.getRegionsInTransition();
|
||||
for (RegionState regionState : rits.values()) {
|
||||
Set<RegionState> rits = regionStates.getRegionsInTransition();
|
||||
for (RegionState regionState : rits) {
|
||||
LOG.info("Processing " + regionState);
|
||||
ServerName serverName = regionState.getServerName();
|
||||
// Server could be null in case of FAILED_OPEN when master cannot find a region plan. In that
|
||||
@ -3309,7 +3309,7 @@ public class AssignmentManager extends ZooKeeperListener {
|
||||
long oldestRITTime = 0;
|
||||
int ritThreshold = this.server.getConfiguration().
|
||||
getInt(HConstants.METRICS_RIT_STUCK_WARNING_THRESHOLD, 60000);
|
||||
for (RegionState state: regionStates.getRegionsInTransition().values()) {
|
||||
for (RegionState state: regionStates.getRegionsInTransition()) {
|
||||
totalRITs++;
|
||||
long ritTime = currentTime - state.getStamp();
|
||||
if (ritTime > ritThreshold) { // more than the threshold
|
||||
|
@ -1266,7 +1266,7 @@ public class HMaster extends HRegionServer implements MasterServices, Server {
|
||||
if (!this.loadBalancerTracker.isBalancerOn()) return false;
|
||||
// Only allow one balance run at at time.
|
||||
if (this.assignmentManager.getRegionStates().isRegionsInTransition()) {
|
||||
Map<String, RegionState> regionsInTransition =
|
||||
Set<RegionState> regionsInTransition =
|
||||
this.assignmentManager.getRegionStates().getRegionsInTransition();
|
||||
// if hbase:meta region is in transition, result of assignment cannot be recorded
|
||||
// ignore the force flag in that case
|
||||
@ -2151,7 +2151,7 @@ public class HMaster extends HRegionServer implements MasterServices, Server {
|
||||
|
||||
String clusterId = fileSystemManager != null ?
|
||||
fileSystemManager.getClusterId().toString() : null;
|
||||
Map<String, RegionState> regionsInTransition = assignmentManager != null ?
|
||||
Set<RegionState> regionsInTransition = assignmentManager != null ?
|
||||
assignmentManager.getRegionStates().getRegionsInTransition() : null;
|
||||
String[] coprocessors = cpHost != null ? getMasterCoprocessors() : null;
|
||||
boolean balancerOn = loadBalancerTracker != null ?
|
||||
|
@ -24,6 +24,7 @@ import java.io.PrintStream;
|
||||
import java.io.PrintWriter;
|
||||
import java.util.Date;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
|
||||
import javax.servlet.http.HttpServletRequest;
|
||||
import javax.servlet.http.HttpServletResponse;
|
||||
@ -116,11 +117,9 @@ public class MasterDumpServlet extends StateDumpServlet {
|
||||
return;
|
||||
}
|
||||
|
||||
Map<String, RegionState> regionsInTransition =
|
||||
am.getRegionStates().getRegionsInTransition();
|
||||
for (Map.Entry<String, RegionState> e : regionsInTransition.entrySet()) {
|
||||
String rid = e.getKey();
|
||||
RegionState rs = e.getValue();
|
||||
Set<RegionState> regionsInTransition = am.getRegionStates().getRegionsInTransition();
|
||||
for (RegionState rs : regionsInTransition) {
|
||||
String rid = rs.getRegion().getRegionNameAsString();
|
||||
out.println("Region " + rid + ": " + rs.toDescriptiveString());
|
||||
}
|
||||
}
|
||||
|
@ -28,6 +28,7 @@ import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.TreeMap;
|
||||
import java.util.TreeSet;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
import org.apache.commons.logging.LogFactory;
|
||||
@ -208,9 +209,8 @@ public class RegionStates {
|
||||
/**
|
||||
* Get regions in transition and their states
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
public synchronized Map<String, RegionState> getRegionsInTransition() {
|
||||
return (Map<String, RegionState>)regionsInTransition.clone();
|
||||
public synchronized Set<RegionState> getRegionsInTransition() {
|
||||
return new HashSet<RegionState>(regionsInTransition.values());
|
||||
}
|
||||
|
||||
/**
|
||||
|
@ -554,10 +554,10 @@ public class HBaseFsck extends Configured implements Closeable {
|
||||
errors.print("Number of requests: " + status.getRequestsCount());
|
||||
errors.print("Number of regions: " + status.getRegionsCount());
|
||||
|
||||
Map<String, RegionState> rits = status.getRegionsInTransition();
|
||||
Set<RegionState> rits = status.getRegionsInTransition();
|
||||
errors.print("Number of regions in transition: " + rits.size());
|
||||
if (details) {
|
||||
for (RegionState state: rits.values()) {
|
||||
for (RegionState state: rits) {
|
||||
errors.print(" " + state.toDescriptiveString());
|
||||
}
|
||||
}
|
||||
@ -732,7 +732,7 @@ public class HBaseFsck extends Configured implements Closeable {
|
||||
checkAndFixOrphanedTableZNodes();
|
||||
|
||||
checkAndFixReplication();
|
||||
|
||||
|
||||
// Remove the hbck lock
|
||||
unlockHbck();
|
||||
|
||||
@ -4217,7 +4217,7 @@ public class HBaseFsck extends Configured implements Closeable {
|
||||
public boolean shouldDisableSplitAndMerge() {
|
||||
return fixAny || disableSplitAndMerge;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Set summary mode.
|
||||
* Print only summary of the tables and status (OK or INCONSISTENT)
|
||||
@ -4249,7 +4249,7 @@ public class HBaseFsck extends Configured implements Closeable {
|
||||
fixTableLocks = shouldFix;
|
||||
fixAny |= shouldFix;
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Set replication fix mode.
|
||||
*/
|
||||
@ -4534,7 +4534,7 @@ public class HBaseFsck extends Configured implements Closeable {
|
||||
out.println("");
|
||||
out.println(" Replication options");
|
||||
out.println(" -fixReplication Deletes replication queues for removed peers");
|
||||
|
||||
|
||||
out.flush();
|
||||
errors.reportError(ERROR_CODE.WRONG_USAGE, sw.toString());
|
||||
|
||||
|
@ -130,10 +130,14 @@ public class HBaseFsckRepair {
|
||||
long expiration = timeout + EnvironmentEdgeManager.currentTime();
|
||||
while (EnvironmentEdgeManager.currentTime() < expiration) {
|
||||
try {
|
||||
Map<String, RegionState> rits=
|
||||
admin.getClusterStatus().getRegionsInTransition();
|
||||
|
||||
if (rits.keySet() != null && !rits.keySet().contains(region.getEncodedName())) {
|
||||
boolean inTransition = false;
|
||||
for (RegionState rs: admin.getClusterStatus().getRegionsInTransition()) {
|
||||
if (rs.getRegion().equals(region)) {
|
||||
inTransition = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (!inTransition) {
|
||||
// yay! no longer RIT
|
||||
return;
|
||||
}
|
||||
|
@ -255,10 +255,9 @@ public class TestDrainingServer {
|
||||
|
||||
am.assign(bulk);
|
||||
|
||||
Map<String, RegionState> regionsInTransition = am.getRegionStates().getRegionsInTransition();
|
||||
for (Entry<String, RegionState> entry : regionsInTransition.entrySet()) {
|
||||
setRegionOpenedOnZK(zkWatcher, entry.getValue().getServerName(),
|
||||
entry.getValue().getRegion());
|
||||
Set<RegionState> regionsInTransition = am.getRegionStates().getRegionsInTransition();
|
||||
for (RegionState rs : regionsInTransition) {
|
||||
setRegionOpenedOnZK(zkWatcher, rs.getServerName(), rs.getRegion());
|
||||
}
|
||||
|
||||
am.waitForAssignment(REGIONINFO_A);
|
||||
|
@ -30,6 +30,7 @@ import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.NavigableMap;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
@ -851,7 +852,7 @@ public class TestMasterObserver {
|
||||
public boolean wasSnapshotCalled() {
|
||||
return preSnapshotCalled && postSnapshotCalled;
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void preListSnapshot(final ObserverContext<MasterCoprocessorEnvironment> ctx,
|
||||
final SnapshotDescription snapshot) throws IOException {
|
||||
@ -1201,7 +1202,7 @@ public class TestMasterObserver {
|
||||
public void postTableFlush(ObserverContext<MasterCoprocessorEnvironment> ctx,
|
||||
TableName tableName) throws IOException {
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void preSetUserQuota(final ObserverContext<MasterCoprocessorEnvironment> ctx,
|
||||
final String userName, final Quotas quotas) throws IOException {
|
||||
@ -1501,7 +1502,7 @@ public class TestMasterObserver {
|
||||
admin.snapshot(TEST_SNAPSHOT, tableName);
|
||||
assertTrue("Coprocessor should have been called on snapshot",
|
||||
cp.wasSnapshotCalled());
|
||||
|
||||
|
||||
//Test list operation
|
||||
admin.listSnapshots();
|
||||
assertTrue("Coprocessor should have been called on snapshot list",
|
||||
@ -1681,8 +1682,7 @@ public class TestMasterObserver {
|
||||
|
||||
// wait for assignments to finish, if any
|
||||
AssignmentManager mgr = master.getAssignmentManager();
|
||||
Collection<RegionState> transRegions =
|
||||
mgr.getRegionStates().getRegionsInTransition().values();
|
||||
Set<RegionState> transRegions = mgr.getRegionStates().getRegionsInTransition();
|
||||
for (RegionState state : transRegions) {
|
||||
mgr.getRegionStates().waitOnRegionToClearRegionsInTransition(state.getRegion());
|
||||
}
|
||||
@ -1718,8 +1718,7 @@ public class TestMasterObserver {
|
||||
private void waitForRITtoBeZero(HMaster master) throws Exception {
|
||||
// wait for assignments to finish
|
||||
AssignmentManager mgr = master.getAssignmentManager();
|
||||
Collection<RegionState> transRegions =
|
||||
mgr.getRegionStates().getRegionsInTransition().values();
|
||||
Set<RegionState> transRegions = mgr.getRegionStates().getRegionsInTransition();
|
||||
for (RegionState state : transRegions) {
|
||||
mgr.getRegionStates().waitOnRegionToClearRegionsInTransition(state.getRegion());
|
||||
}
|
||||
|
@ -51,27 +51,6 @@ public class Mocking {
|
||||
}
|
||||
}
|
||||
|
||||
static void waitForRegionPendingOpenInRIT(AssignmentManager am, String encodedName)
|
||||
throws InterruptedException {
|
||||
// We used to do a check like this:
|
||||
//!Mocking.verifyRegionState(this.watcher, REGIONINFO, EventType.M_ZK_REGION_OFFLINE)) {
|
||||
// There is a race condition with this: because we may do the transition to
|
||||
// RS_ZK_REGION_OPENING before the RIT is internally updated. We need to wait for the
|
||||
// RIT to be as we need it to be instead. This cannot happen in a real cluster as we
|
||||
// update the RIT before sending the openRegion request.
|
||||
|
||||
boolean wait = true;
|
||||
while (wait) {
|
||||
RegionState state = am.getRegionStates()
|
||||
.getRegionsInTransition().get(encodedName);
|
||||
if (state != null && state.isPendingOpen()){
|
||||
wait = false;
|
||||
} else {
|
||||
Thread.sleep(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Verifies that the specified region is in the specified state in ZooKeeper.
|
||||
* <p>
|
||||
|
@ -24,7 +24,6 @@ import java.io.IOException;
|
||||
import java.io.StringWriter;
|
||||
import java.util.HashSet;
|
||||
import java.util.List;
|
||||
import java.util.NavigableMap;
|
||||
import java.util.Set;
|
||||
import java.util.regex.Matcher;
|
||||
import java.util.regex.Pattern;
|
||||
@ -85,10 +84,8 @@ public class TestMasterStatusServlet {
|
||||
// Fake AssignmentManager and RIT
|
||||
AssignmentManager am = Mockito.mock(AssignmentManager.class);
|
||||
RegionStates rs = Mockito.mock(RegionStates.class);
|
||||
NavigableMap<String, RegionState> regionsInTransition =
|
||||
Maps.newTreeMap();
|
||||
regionsInTransition.put("r1",
|
||||
new RegionState(FAKE_HRI, RegionState.State.CLOSING, 12345L, FAKE_HOST));
|
||||
Set<RegionState> regionsInTransition = new HashSet<RegionState>();
|
||||
regionsInTransition.add(new RegionState(FAKE_HRI, RegionState.State.CLOSING, 12345L, FAKE_HOST));
|
||||
Mockito.doReturn(rs).when(am).getRegionStates();
|
||||
Mockito.doReturn(regionsInTransition).when(rs).getRegionsInTransition();
|
||||
Mockito.doReturn(am).when(master).getAssignmentManager();
|
||||
@ -160,19 +157,17 @@ public class TestMasterStatusServlet {
|
||||
RegionStates rs = Mockito.mock(RegionStates.class);
|
||||
|
||||
// Add 100 regions as in-transition
|
||||
NavigableMap<String, RegionState> regionsInTransition =
|
||||
Maps.newTreeMap();
|
||||
HashSet<RegionState> regionsInTransition = new HashSet<RegionState>();
|
||||
for (byte i = 0; i < 100; i++) {
|
||||
HRegionInfo hri = new HRegionInfo(FAKE_TABLE.getTableName(),
|
||||
new byte[]{i}, new byte[]{(byte) (i+1)});
|
||||
regionsInTransition.put(hri.getEncodedName(),
|
||||
regionsInTransition.add(
|
||||
new RegionState(hri, RegionState.State.CLOSING, 12345L, FAKE_HOST));
|
||||
}
|
||||
// Add hbase:meta in transition as well
|
||||
regionsInTransition.put(
|
||||
HRegionInfo.FIRST_META_REGIONINFO.getEncodedName(),
|
||||
regionsInTransition.add(
|
||||
new RegionState(HRegionInfo.FIRST_META_REGIONINFO,
|
||||
RegionState.State.CLOSING, 12345L, FAKE_HOST));
|
||||
RegionState.State.CLOSING, 123L, FAKE_HOST));
|
||||
Mockito.doReturn(rs).when(am).getRegionStates();
|
||||
Mockito.doReturn(regionsInTransition).when(rs).getRegionsInTransition();
|
||||
|
||||
|
@ -31,6 +31,7 @@ import java.io.InterruptedIOException;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
|
||||
import org.apache.commons.logging.Log;
|
||||
@ -257,13 +258,10 @@ public class TestSplitTransactionOnCluster {
|
||||
assertTrue("fist split did not complete", firstSplitCompleted);
|
||||
|
||||
RegionStates regionStates = cluster.getMaster().getAssignmentManager().getRegionStates();
|
||||
Map<String, RegionState> rit = regionStates.getRegionsInTransition();
|
||||
|
||||
for (int i=0; rit.containsKey(hri.getTable()) && i<100; i++) {
|
||||
for (int i = 0; regionStates.isRegionInTransition(hri) && i < 100; i++) {
|
||||
Thread.sleep(100);
|
||||
}
|
||||
assertFalse("region still in transition", rit.containsKey(
|
||||
rit.containsKey(hri.getTable())));
|
||||
assertFalse("region still in transition", regionStates.isRegionInTransition(hri));
|
||||
|
||||
List<Region> onlineRegions = regionServer.getOnlineRegions(tableName);
|
||||
// Region server side split is successful.
|
||||
@ -322,8 +320,7 @@ public class TestSplitTransactionOnCluster {
|
||||
@Override
|
||||
public boolean evaluate() throws Exception {
|
||||
RegionStates regionStates = cluster.getMaster().getAssignmentManager().getRegionStates();
|
||||
Map<String, RegionState> rit = regionStates.getRegionsInTransition();
|
||||
return (rit.size() == 0);
|
||||
return !regionStates.isRegionsInTransition();
|
||||
}
|
||||
});
|
||||
} finally {
|
||||
@ -383,8 +380,7 @@ public class TestSplitTransactionOnCluster {
|
||||
@Override
|
||||
public boolean evaluate() throws Exception {
|
||||
RegionStates regionStates = cluster.getMaster().getAssignmentManager().getRegionStates();
|
||||
Map<String, RegionState> rit = regionStates.getRegionsInTransition();
|
||||
return (rit.size() == 0);
|
||||
return !regionStates.isRegionsInTransition();
|
||||
}
|
||||
});
|
||||
assertEquals(2, cluster.getRegions(tableName).size());
|
||||
@ -954,7 +950,7 @@ public class TestSplitTransactionOnCluster {
|
||||
tableExists = MetaTableAccessor.tableExists(regionServer.getConnection(),
|
||||
tableName);
|
||||
assertEquals("The specified table should present.", true, tableExists);
|
||||
Map<String, RegionState> rit = cluster.getMaster().getAssignmentManager().getRegionStates()
|
||||
Set<RegionState> rit = cluster.getMaster().getAssignmentManager().getRegionStates()
|
||||
.getRegionsInTransition();
|
||||
assertTrue(rit.size() == 3);
|
||||
cluster.getMaster().getAssignmentManager().regionOffline(st.getFirstDaughter());
|
||||
@ -1329,14 +1325,13 @@ public class TestSplitTransactionOnCluster {
|
||||
@Override
|
||||
public boolean evaluate() throws Exception {
|
||||
RegionStates regionStates = cluster.getMaster().getAssignmentManager().getRegionStates();
|
||||
Map<String, RegionState> rit = regionStates.getRegionsInTransition();
|
||||
return (rit.size() == 0);
|
||||
return !regionStates.isRegionsInTransition();
|
||||
}
|
||||
});
|
||||
regions = TESTING_UTIL.getHBaseAdmin().getTableRegions(tableName);
|
||||
assertTrue(regions.size() == 1);
|
||||
RegionStates regionStates = cluster.getMaster().getAssignmentManager().getRegionStates();
|
||||
Map<String, RegionState> rit = regionStates.getRegionsInTransition();
|
||||
Set<RegionState> rit = regionStates.getRegionsInTransition();
|
||||
assertTrue(rit.size() == 0);
|
||||
} finally {
|
||||
table.close();
|
||||
|
@ -472,7 +472,7 @@ public class TestHBaseFsck {
|
||||
i++;
|
||||
}
|
||||
return i;
|
||||
}
|
||||
}
|
||||
/**
|
||||
* delete table in preparation for next test
|
||||
*
|
||||
@ -1251,13 +1251,13 @@ public class TestHBaseFsck {
|
||||
try {
|
||||
setupTable(table);
|
||||
assertEquals(ROWKEYS.length, countRows());
|
||||
|
||||
|
||||
// Mess it up by leaving a hole in the meta data
|
||||
admin.disableTable(table);
|
||||
deleteRegion(conf, tbl.getTableDescriptor(), Bytes.toBytes("B"), Bytes.toBytes("C"), true,
|
||||
true, false, true, HRegionInfo.DEFAULT_REPLICA_ID);
|
||||
admin.enableTable(table);
|
||||
|
||||
|
||||
HBaseFsck hbck = doFsck(conf, false);
|
||||
assertErrors(hbck,
|
||||
new HBaseFsck.ErrorReporter.ERROR_CODE[] {
|
||||
@ -1266,13 +1266,13 @@ public class TestHBaseFsck {
|
||||
HBaseFsck.ErrorReporter.ERROR_CODE.HOLE_IN_REGION_CHAIN });
|
||||
// holes are separate from overlap groups
|
||||
assertEquals(0, hbck.getOverlapGroups(table).size());
|
||||
|
||||
|
||||
// fix hole
|
||||
doFsck(conf, true);
|
||||
|
||||
|
||||
// check that hole fixed
|
||||
assertNoErrors(doFsck(conf, false));
|
||||
|
||||
|
||||
// check data belong to the correct region,every scan should get one row.
|
||||
for (int i = 0; i < ROWKEYS.length; i++) {
|
||||
if (i != ROWKEYS.length - 1) {
|
||||
@ -1281,12 +1281,12 @@ public class TestHBaseFsck {
|
||||
assertEquals(1, countRows(ROWKEYS[i], null));
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
} finally {
|
||||
cleanupTable(table);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* This creates and fixes a bad table with a region that is missing meta and
|
||||
* not assigned to a region server.
|
||||
@ -2268,13 +2268,13 @@ public class TestHBaseFsck {
|
||||
doQuarantineTest(table, hbck, 3, 0, 0, 0, 1);
|
||||
hbck.close();
|
||||
}
|
||||
|
||||
|
||||
@Test(timeout=60000)
|
||||
public void testCheckReplication() throws Exception {
|
||||
// check no errors
|
||||
HBaseFsck hbck = doFsck(conf, false);
|
||||
assertNoErrors(hbck);
|
||||
|
||||
|
||||
// create peer
|
||||
ReplicationAdmin replicationAdmin = new ReplicationAdmin(conf);
|
||||
Assert.assertEquals(0, replicationAdmin.getPeersCount());
|
||||
@ -2282,7 +2282,7 @@ public class TestHBaseFsck {
|
||||
replicationAdmin.addPeer("1", "127.0.0.1:2181" + zkPort + ":/hbase");
|
||||
replicationAdmin.getPeersCount();
|
||||
Assert.assertEquals(1, replicationAdmin.getPeersCount());
|
||||
|
||||
|
||||
// create replicator
|
||||
ZooKeeperWatcher zkw = new ZooKeeperWatcher(conf, "Test Hbase Fsck", connection);
|
||||
ReplicationQueues repQueues =
|
||||
@ -2294,7 +2294,7 @@ public class TestHBaseFsck {
|
||||
Assert.assertEquals(2, repQueues.getAllQueues().size());
|
||||
hbck = doFsck(conf, false);
|
||||
assertNoErrors(hbck);
|
||||
|
||||
|
||||
// queues for removed peer
|
||||
repQueues.addLog("2", "file1");
|
||||
repQueues.addLog("2-server2", "file1");
|
||||
@ -2302,7 +2302,7 @@ public class TestHBaseFsck {
|
||||
hbck = doFsck(conf, false);
|
||||
assertErrors(hbck, new ERROR_CODE[] { ERROR_CODE.UNDELETED_REPLICATION_QUEUE,
|
||||
ERROR_CODE.UNDELETED_REPLICATION_QUEUE });
|
||||
|
||||
|
||||
// fix the case
|
||||
hbck = doFsck(conf, true);
|
||||
hbck = doFsck(conf, false);
|
||||
@ -2311,7 +2311,7 @@ public class TestHBaseFsck {
|
||||
Assert.assertEquals(2, repQueues.getAllQueues().size());
|
||||
Assert.assertNull(repQueues.getLogsInQueue("2"));
|
||||
Assert.assertNull(repQueues.getLogsInQueue("2-sever2"));
|
||||
|
||||
|
||||
replicationAdmin.removePeer("1");
|
||||
repQueues.removeAllQueues();
|
||||
zkw.close();
|
||||
@ -2841,8 +2841,8 @@ public class TestHBaseFsck {
|
||||
st.prepare();
|
||||
st.stepsBeforePONR(regionServer, regionServer, false);
|
||||
AssignmentManager am = cluster.getMaster().getAssignmentManager();
|
||||
Map<String, RegionState> regionsInTransition = am.getRegionStates().getRegionsInTransition();
|
||||
for (RegionState state : regionsInTransition.values()) {
|
||||
Set<RegionState> regionsInTransition = am.getRegionStates().getRegionsInTransition();
|
||||
for (RegionState state : regionsInTransition) {
|
||||
am.regionOffline(state.getRegion());
|
||||
}
|
||||
ZKAssign.deleteNodeFailSilent(regionServer.getZooKeeper(), regions.get(0).getRegionInfo());
|
||||
|
@ -679,7 +679,7 @@ module Hbase
|
||||
puts("version %s" % [ status.getHBaseVersion() ])
|
||||
# Put regions in transition first because usually empty
|
||||
puts("%d regionsInTransition" % status.getRegionsInTransition().size())
|
||||
for k, v in status.getRegionsInTransition()
|
||||
for v in status.getRegionsInTransition()
|
||||
puts(" %s" % [v])
|
||||
end
|
||||
master = status.getMaster()
|
||||
|
Loading…
x
Reference in New Issue
Block a user