HBASE-27250 MasterRpcService#setRegionStateInMeta does not support replica region encodedNames or region names
- Added sanity check to make sure input region encoded name or region name is valid - Assignment improvements pertaining to read replica regions - make several MetaTableAccessor methods more precise in their handling of replica regions - hbck2 setRegionStateInMeta and HBCKServerCrashProcedure handle read replicas - separate AM helper methods -- loading RegionInfo from cache vs. refreshing cache from meta - AM helper method support loading RegionInfo from cache via either region name and encoded region name (both caches are maintained, and under lock) - consolidate, extend tests to cover read replica regions Co-authored-by: Huaxiang Sun <huaxiangsun@apache.com> Co-authored-by: Nick Dimiduk <ndimiduk@apache.org> Signed-off-by: Peter Somogyi <psomogyi@apache.org>
This commit is contained in:
huaxiangsun
GitHub
parent
cf179d3fb9
commit
22dbb7afc3
|
|
@ -192,7 +192,8 @@ public final class MetaTableAccessor {
|
|||
RegionLocations locations = CatalogFamilyFormat.getRegionLocations(r);
|
||||
return locations == null
|
||||
? null
|
||||
: locations.getRegionLocation(parsedInfo == null ? 0 : parsedInfo.getReplicaId());
|
||||
: locations.getRegionLocation(
|
||||
parsedInfo == null ? RegionInfo.DEFAULT_REPLICA_ID : parsedInfo.getReplicaId());
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -220,12 +221,12 @@ public final class MetaTableAccessor {
|
|||
/**
|
||||
* Gets the result in hbase:meta for the specified region.
|
||||
* @param connection connection we're using
|
||||
* @param regionName region we're looking for
|
||||
* @param regionInfo region we're looking for
|
||||
* @return result of the specified region
|
||||
*/
|
||||
public static Result getRegionResult(Connection connection, byte[] regionName)
|
||||
public static Result getRegionResult(Connection connection, RegionInfo regionInfo)
|
||||
throws IOException {
|
||||
Get get = new Get(regionName);
|
||||
Get get = new Get(CatalogFamilyFormat.getMetaKeyForRegion(regionInfo));
|
||||
get.addFamily(HConstants.CATALOG_FAMILY);
|
||||
try (Table t = getMetaHTable(connection)) {
|
||||
return t.get(get);
|
||||
|
|
@ -621,11 +622,22 @@ public final class MetaTableAccessor {
|
|||
////////////////////////
|
||||
// Editing operations //
|
||||
////////////////////////
|
||||
|
||||
/**
|
||||
* Generates and returns a Put containing the region into for the catalog table
|
||||
* Generates and returns a {@link Put} containing the {@link RegionInfo} for the catalog table.
|
||||
* @throws IllegalArgumentException when the provided RegionInfo is not the default replica.
|
||||
*/
|
||||
public static Put makePutFromRegionInfo(RegionInfo regionInfo) throws IOException {
|
||||
return makePutFromRegionInfo(regionInfo, EnvironmentEdgeManager.currentTime());
|
||||
}
|
||||
|
||||
/**
|
||||
* Generates and returns a {@link Put} containing the {@link RegionInfo} for the catalog table.
|
||||
* @throws IllegalArgumentException when the provided RegionInfo is not the default replica.
|
||||
*/
|
||||
public static Put makePutFromRegionInfo(RegionInfo regionInfo, long ts) throws IOException {
|
||||
return addRegionInfo(new Put(regionInfo.getRegionName(), ts), regionInfo);
|
||||
return addRegionInfo(new Put(CatalogFamilyFormat.getMetaKeyForRegion(regionInfo), ts),
|
||||
regionInfo);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -635,7 +647,11 @@ public final class MetaTableAccessor {
|
|||
if (regionInfo == null) {
|
||||
throw new IllegalArgumentException("Can't make a delete for null region");
|
||||
}
|
||||
Delete delete = new Delete(regionInfo.getRegionName());
|
||||
if (regionInfo.getReplicaId() != RegionInfo.DEFAULT_REPLICA_ID) {
|
||||
throw new IllegalArgumentException(
|
||||
"Can't make delete for a replica region. Operate on the primary");
|
||||
}
|
||||
Delete delete = new Delete(CatalogFamilyFormat.getMetaKeyForRegion(regionInfo));
|
||||
delete.addFamily(HConstants.CATALOG_FAMILY, ts);
|
||||
return delete;
|
||||
}
|
||||
|
|
@ -726,9 +742,15 @@ public final class MetaTableAccessor {
|
|||
}
|
||||
}
|
||||
|
||||
public static Put addRegionStateToPut(Put put, RegionState.State state) throws IOException {
|
||||
/**
|
||||
* Set the column value corresponding to this {@code replicaId}'s {@link RegionState} to the
|
||||
* provided {@code state}. Mutates the provided {@link Put}.
|
||||
*/
|
||||
public static Put addRegionStateToPut(Put put, int replicaId, RegionState.State state)
|
||||
throws IOException {
|
||||
put.add(CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY).setRow(put.getRow())
|
||||
.setFamily(HConstants.CATALOG_FAMILY).setQualifier(HConstants.STATE_QUALIFIER)
|
||||
.setFamily(HConstants.CATALOG_FAMILY)
|
||||
.setQualifier(CatalogFamilyFormat.getRegionStateColumn(replicaId))
|
||||
.setTimestamp(put.getTimestamp()).setType(Cell.Type.Put).setValue(Bytes.toBytes(state.name()))
|
||||
.build());
|
||||
return put;
|
||||
|
|
@ -739,8 +761,9 @@ public final class MetaTableAccessor {
|
|||
*/
|
||||
public static void updateRegionState(Connection connection, RegionInfo ri,
|
||||
RegionState.State state) throws IOException {
|
||||
Put put = new Put(RegionReplicaUtil.getRegionInfoForDefaultReplica(ri).getRegionName());
|
||||
putsToMetaTable(connection, Collections.singletonList(addRegionStateToPut(put, state)));
|
||||
final Put put = makePutFromRegionInfo(ri);
|
||||
addRegionStateToPut(put, ri.getReplicaId(), state);
|
||||
putsToMetaTable(connection, Collections.singletonList(put));
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -759,7 +782,7 @@ public final class MetaTableAccessor {
|
|||
public static void addSplitsToParent(Connection connection, RegionInfo regionInfo,
|
||||
RegionInfo splitA, RegionInfo splitB) throws IOException {
|
||||
try (Table meta = getMetaHTable(connection)) {
|
||||
Put put = makePutFromRegionInfo(regionInfo, EnvironmentEdgeManager.currentTime());
|
||||
Put put = makePutFromRegionInfo(regionInfo);
|
||||
addDaughtersToPut(put, splitA, splitB);
|
||||
meta.put(put);
|
||||
debugLogMutation(put);
|
||||
|
|
@ -797,7 +820,7 @@ public final class MetaTableAccessor {
|
|||
}
|
||||
Put put = makePutFromRegionInfo(regionInfo, ts);
|
||||
// New regions are added with initial state of CLOSED.
|
||||
addRegionStateToPut(put, RegionState.State.CLOSED);
|
||||
addRegionStateToPut(put, regionInfo.getReplicaId(), RegionState.State.CLOSED);
|
||||
// Add empty locations for region replicas so that number of replicas can be cached
|
||||
// whenever the primary region is looked up from meta
|
||||
for (int i = 1; i < regionReplication; i++) {
|
||||
|
|
|
|||
|
|
@ -28,6 +28,7 @@ import java.util.HashSet;
|
|||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.Optional;
|
||||
import java.util.Set;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
|
|
@ -49,7 +50,6 @@ import org.apache.hadoop.hbase.client.BalanceRequest;
|
|||
import org.apache.hadoop.hbase.client.BalanceResponse;
|
||||
import org.apache.hadoop.hbase.client.MasterSwitchType;
|
||||
import org.apache.hadoop.hbase.client.NormalizeTableFilterParams;
|
||||
import org.apache.hadoop.hbase.client.Put;
|
||||
import org.apache.hadoop.hbase.client.RegionInfo;
|
||||
import org.apache.hadoop.hbase.client.Table;
|
||||
import org.apache.hadoop.hbase.client.TableDescriptor;
|
||||
|
|
@ -66,6 +66,7 @@ import org.apache.hadoop.hbase.ipc.RpcServer;
|
|||
import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface;
|
||||
import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
|
||||
import org.apache.hadoop.hbase.ipc.ServerRpcController;
|
||||
import org.apache.hadoop.hbase.master.assignment.AssignmentManager;
|
||||
import org.apache.hadoop.hbase.master.assignment.RegionStateNode;
|
||||
import org.apache.hadoop.hbase.master.assignment.RegionStates;
|
||||
import org.apache.hadoop.hbase.master.cleaner.HFileCleaner;
|
||||
|
|
@ -2549,64 +2550,69 @@ public class MasterRpcServices extends HBaseRpcServicesBase<HMaster>
|
|||
SetRegionStateInMetaRequest request) throws ServiceException {
|
||||
rpcPreCheck("setRegionStateInMeta");
|
||||
SetRegionStateInMetaResponse.Builder builder = SetRegionStateInMetaResponse.newBuilder();
|
||||
final AssignmentManager am = server.getAssignmentManager();
|
||||
try {
|
||||
for (RegionSpecifierAndState s : request.getStatesList()) {
|
||||
RegionSpecifier spec = s.getRegionSpecifier();
|
||||
String encodedName;
|
||||
if (spec.getType() == RegionSpecifierType.ENCODED_REGION_NAME) {
|
||||
encodedName = spec.getValue().toStringUtf8();
|
||||
} else {
|
||||
// TODO: actually, a full region name can save a lot on meta scan, improve later.
|
||||
encodedName = RegionInfo.encodeRegionName(spec.getValue().toByteArray());
|
||||
final RegionSpecifier spec = s.getRegionSpecifier();
|
||||
final RegionInfo targetRegionInfo = getRegionInfo(spec);
|
||||
final RegionState.State targetState = RegionState.State.convert(s.getState());
|
||||
final RegionState.State currentState = Optional.ofNullable(targetRegionInfo)
|
||||
.map(info -> am.getRegionStates().getRegionState(info)).map(RegionState::getState)
|
||||
.orElseThrow(
|
||||
() -> new ServiceException("No existing state known for region '" + spec + "'."));
|
||||
LOG.info("{} set region={} state from {} to {}", server.getClientIdAuditPrefix(),
|
||||
targetRegionInfo, currentState, targetState);
|
||||
if (currentState == targetState) {
|
||||
LOG.debug("Proposed state matches current state. {}, {}", targetRegionInfo, currentState);
|
||||
continue;
|
||||
}
|
||||
RegionInfo info = this.server.getAssignmentManager().loadRegionFromMeta(encodedName);
|
||||
LOG.trace("region info loaded from meta table: {}", info);
|
||||
RegionState prevState =
|
||||
this.server.getAssignmentManager().getRegionStates().getRegionState(info);
|
||||
RegionState.State newState = RegionState.State.convert(s.getState());
|
||||
LOG.info("{} set region={} state from {} to {}", server.getClientIdAuditPrefix(), info,
|
||||
prevState.getState(), newState);
|
||||
Put metaPut =
|
||||
MetaTableAccessor.makePutFromRegionInfo(info, EnvironmentEdgeManager.currentTime());
|
||||
metaPut.addColumn(HConstants.CATALOG_FAMILY, HConstants.STATE_QUALIFIER,
|
||||
Bytes.toBytes(newState.name()));
|
||||
List<Put> putList = new ArrayList<>();
|
||||
putList.add(metaPut);
|
||||
MetaTableAccessor.putsToMetaTable(this.server.getConnection(), putList);
|
||||
MetaTableAccessor.updateRegionState(server.getConnection(), targetRegionInfo, targetState);
|
||||
// Loads from meta again to refresh AM cache with the new region state
|
||||
this.server.getAssignmentManager().loadRegionFromMeta(encodedName);
|
||||
am.populateRegionStatesFromMeta(targetRegionInfo);
|
||||
builder.addStates(RegionSpecifierAndState.newBuilder().setRegionSpecifier(spec)
|
||||
.setState(prevState.getState().convert()));
|
||||
.setState(currentState.convert()));
|
||||
}
|
||||
} catch (Exception e) {
|
||||
} catch (IOException e) {
|
||||
throw new ServiceException(e);
|
||||
}
|
||||
return builder.build();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get RegionInfo from Master using content of RegionSpecifier as key.
|
||||
* @return RegionInfo found by decoding <code>rs</code> or null if none found
|
||||
* Get {@link RegionInfo} from Master using content of {@link RegionSpecifier} as key.
|
||||
* @return {@link RegionInfo} found by decoding {@code rs} or {@code null} if {@code rs} is
|
||||
* unknown to the master.
|
||||
* @throws ServiceException If some error occurs while querying META or parsing results.
|
||||
*/
|
||||
private RegionInfo getRegionInfo(HBaseProtos.RegionSpecifier rs) throws UnknownRegionException {
|
||||
RegionInfo ri = null;
|
||||
private RegionInfo getRegionInfo(HBaseProtos.RegionSpecifier rs) throws ServiceException {
|
||||
// TODO: this doesn't handle MOB regions. Should it? See the public method #getRegionInfo
|
||||
final AssignmentManager am = server.getAssignmentManager();
|
||||
final String encodedRegionName;
|
||||
final RegionInfo info;
|
||||
// first try resolving from the AM's caches.
|
||||
switch (rs.getType()) {
|
||||
case REGION_NAME:
|
||||
final byte[] regionName = rs.getValue().toByteArray();
|
||||
ri = this.server.getAssignmentManager().getRegionInfo(regionName);
|
||||
encodedRegionName = RegionInfo.encodeRegionName(regionName);
|
||||
info = am.getRegionInfo(regionName);
|
||||
break;
|
||||
case ENCODED_REGION_NAME:
|
||||
String encodedRegionName = Bytes.toString(rs.getValue().toByteArray());
|
||||
RegionState regionState =
|
||||
this.server.getAssignmentManager().getRegionStates().getRegionState(encodedRegionName);
|
||||
ri = regionState == null
|
||||
? this.server.getAssignmentManager().loadRegionFromMeta(encodedRegionName)
|
||||
: regionState.getRegion();
|
||||
encodedRegionName = rs.getValue().toStringUtf8();
|
||||
info = am.getRegionInfo(encodedRegionName);
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
throw new IllegalArgumentException("Unrecognized RegionSpecifierType " + rs.getType());
|
||||
}
|
||||
return ri;
|
||||
if (info != null) {
|
||||
return info;
|
||||
}
|
||||
// fall back to a meta scan and check the cache again.
|
||||
try {
|
||||
am.populateRegionStatesFromMeta(encodedRegionName);
|
||||
} catch (IOException e) {
|
||||
throw new ServiceException(e);
|
||||
}
|
||||
return am.getRegionInfo(encodedRegionName);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -2626,28 +2632,22 @@ public class MasterRpcServices extends HBaseRpcServicesBase<HMaster>
|
|||
public MasterProtos.AssignsResponse assigns(RpcController controller,
|
||||
MasterProtos.AssignsRequest request) throws ServiceException {
|
||||
checkMasterProcedureExecutor();
|
||||
final ProcedureExecutor<MasterProcedureEnv> pe = server.getMasterProcedureExecutor();
|
||||
final AssignmentManager am = server.getAssignmentManager();
|
||||
MasterProtos.AssignsResponse.Builder responseBuilder =
|
||||
MasterProtos.AssignsResponse.newBuilder();
|
||||
try {
|
||||
boolean override = request.getOverride();
|
||||
LOG.info("{} assigns, override={}", server.getClientIdAuditPrefix(), override);
|
||||
for (HBaseProtos.RegionSpecifier rs : request.getRegionList()) {
|
||||
long pid = Procedure.NO_PROC_ID;
|
||||
RegionInfo ri = getRegionInfo(rs);
|
||||
if (ri == null) {
|
||||
LOG.info("Unknown={}", rs);
|
||||
} else {
|
||||
Procedure p = this.server.getAssignmentManager().createOneAssignProcedure(ri, override);
|
||||
if (p != null) {
|
||||
pid = this.server.getMasterProcedureExecutor().submitProcedure(p);
|
||||
}
|
||||
}
|
||||
responseBuilder.addPid(pid);
|
||||
final boolean override = request.getOverride();
|
||||
LOG.info("{} assigns, override={}", server.getClientIdAuditPrefix(), override);
|
||||
for (HBaseProtos.RegionSpecifier rs : request.getRegionList()) {
|
||||
final RegionInfo info = getRegionInfo(rs);
|
||||
if (info == null) {
|
||||
LOG.info("Unknown region {}", rs);
|
||||
continue;
|
||||
}
|
||||
return responseBuilder.build();
|
||||
} catch (IOException ioe) {
|
||||
throw new ServiceException(ioe);
|
||||
responseBuilder.addPid(Optional.ofNullable(am.createOneAssignProcedure(info, override))
|
||||
.map(pe::submitProcedure).orElse(Procedure.NO_PROC_ID));
|
||||
}
|
||||
return responseBuilder.build();
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -2659,35 +2659,29 @@ public class MasterRpcServices extends HBaseRpcServicesBase<HMaster>
|
|||
public MasterProtos.UnassignsResponse unassigns(RpcController controller,
|
||||
MasterProtos.UnassignsRequest request) throws ServiceException {
|
||||
checkMasterProcedureExecutor();
|
||||
final ProcedureExecutor<MasterProcedureEnv> pe = server.getMasterProcedureExecutor();
|
||||
final AssignmentManager am = server.getAssignmentManager();
|
||||
MasterProtos.UnassignsResponse.Builder responseBuilder =
|
||||
MasterProtos.UnassignsResponse.newBuilder();
|
||||
try {
|
||||
boolean override = request.getOverride();
|
||||
LOG.info("{} unassigns, override={}", server.getClientIdAuditPrefix(), override);
|
||||
for (HBaseProtos.RegionSpecifier rs : request.getRegionList()) {
|
||||
long pid = Procedure.NO_PROC_ID;
|
||||
RegionInfo ri = getRegionInfo(rs);
|
||||
if (ri == null) {
|
||||
LOG.info("Unknown={}", rs);
|
||||
} else {
|
||||
Procedure p = this.server.getAssignmentManager().createOneUnassignProcedure(ri, override);
|
||||
if (p != null) {
|
||||
pid = this.server.getMasterProcedureExecutor().submitProcedure(p);
|
||||
}
|
||||
}
|
||||
responseBuilder.addPid(pid);
|
||||
final boolean override = request.getOverride();
|
||||
LOG.info("{} unassigns, override={}", server.getClientIdAuditPrefix(), override);
|
||||
for (HBaseProtos.RegionSpecifier rs : request.getRegionList()) {
|
||||
final RegionInfo info = getRegionInfo(rs);
|
||||
if (info == null) {
|
||||
LOG.info("Unknown region {}", rs);
|
||||
continue;
|
||||
}
|
||||
return responseBuilder.build();
|
||||
} catch (IOException ioe) {
|
||||
throw new ServiceException(ioe);
|
||||
responseBuilder.addPid(Optional.ofNullable(am.createOneUnassignProcedure(info, override))
|
||||
.map(pe::submitProcedure).orElse(Procedure.NO_PROC_ID));
|
||||
}
|
||||
return responseBuilder.build();
|
||||
}
|
||||
|
||||
/**
|
||||
* Bypass specified procedure to completion. Procedure is marked completed but no actual work is
|
||||
* done from the current state/ step onwards. Parents of the procedure are also marked for bypass.
|
||||
* NOTE: this is a dangerous operation and may be used to unstuck buggy procedures. This may leave
|
||||
* system in inconherent state. This may need to be followed by some cleanup steps/ actions by
|
||||
* system in incoherent state. This may need to be followed by some cleanup steps/ actions by
|
||||
* operator.
|
||||
* @return BypassProcedureToCompletionResponse indicating success or failure
|
||||
*/
|
||||
|
|
@ -3370,15 +3364,10 @@ public class MasterRpcServices extends HBaseRpcServicesBase<HMaster>
|
|||
@QosPriority(priority = HConstants.ADMIN_QOS)
|
||||
public GetRegionInfoResponse getRegionInfo(final RpcController controller,
|
||||
final GetRegionInfoRequest request) throws ServiceException {
|
||||
RegionInfo ri = null;
|
||||
try {
|
||||
ri = getRegionInfo(request.getRegion());
|
||||
} catch (UnknownRegionException ure) {
|
||||
throw new ServiceException(ure);
|
||||
}
|
||||
GetRegionInfoResponse.Builder builder = GetRegionInfoResponse.newBuilder();
|
||||
if (ri != null) {
|
||||
builder.setRegionInfo(ProtobufUtil.toRegionInfo(ri));
|
||||
final GetRegionInfoResponse.Builder builder = GetRegionInfoResponse.newBuilder();
|
||||
final RegionInfo info = getRegionInfo(request.getRegion());
|
||||
if (info != null) {
|
||||
builder.setRegionInfo(ProtobufUtil.toRegionInfo(info));
|
||||
} else {
|
||||
// Is it a MOB name? These work differently.
|
||||
byte[] regionName = request.getRegion().getValue().toByteArray();
|
||||
|
|
|
|||
|
|
@ -17,6 +17,7 @@
|
|||
*/
|
||||
package org.apache.hadoop.hbase.master.assignment;
|
||||
|
||||
import edu.umd.cs.findbugs.annotations.NonNull;
|
||||
import java.io.IOException;
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
|
|
@ -1750,26 +1751,29 @@ public class AssignmentManager {
|
|||
};
|
||||
|
||||
/**
|
||||
* Query META if the given <code>RegionInfo</code> exists, adding to
|
||||
* <code>AssignmentManager.regionStateStore</code> cache if the region is found in META.
|
||||
* @param regionEncodedName encoded name for the region to be loaded from META into
|
||||
* <code>AssignmentManager.regionStateStore</code> cache
|
||||
* @return <code>RegionInfo</code> instance for the given region if it is present in META and got
|
||||
* successfully loaded into <code>AssignmentManager.regionStateStore</code> cache,
|
||||
* <b>null</b> otherwise.
|
||||
* @throws UnknownRegionException if any errors occur while querying meta.
|
||||
* Attempt to load {@code regionInfo} from META, adding any results to the
|
||||
* {@link #regionStateStore} Is NOT aware of replica regions.
|
||||
* @param regionInfo the region to be loaded from META.
|
||||
* @throws IOException If some error occurs while querying META or parsing results.
|
||||
*/
|
||||
public RegionInfo loadRegionFromMeta(String regionEncodedName) throws UnknownRegionException {
|
||||
try {
|
||||
RegionMetaLoadingVisitor visitor = new RegionMetaLoadingVisitor();
|
||||
regionStateStore.visitMetaForRegion(regionEncodedName, visitor);
|
||||
return regionStates.getRegionState(regionEncodedName) == null
|
||||
? null
|
||||
: regionStates.getRegionState(regionEncodedName).getRegion();
|
||||
} catch (IOException e) {
|
||||
throw new UnknownRegionException(
|
||||
"Error trying to load region " + regionEncodedName + " from META", e);
|
||||
}
|
||||
public void populateRegionStatesFromMeta(@NonNull final RegionInfo regionInfo)
|
||||
throws IOException {
|
||||
final String regionEncodedName = RegionInfo.DEFAULT_REPLICA_ID == regionInfo.getReplicaId()
|
||||
? regionInfo.getEncodedName()
|
||||
: RegionInfoBuilder.newBuilder(regionInfo).setReplicaId(RegionInfo.DEFAULT_REPLICA_ID).build()
|
||||
.getEncodedName();
|
||||
populateRegionStatesFromMeta(regionEncodedName);
|
||||
}
|
||||
|
||||
/**
|
||||
* Attempt to load {@code regionEncodedName} from META, adding any results to the
|
||||
* {@link #regionStateStore} Is NOT aware of replica regions.
|
||||
* @param regionEncodedName encoded name for the region to be loaded from META.
|
||||
* @throws IOException If some error occurs while querying META or parsing results.
|
||||
*/
|
||||
public void populateRegionStatesFromMeta(@NonNull String regionEncodedName) throws IOException {
|
||||
final RegionMetaLoadingVisitor visitor = new RegionMetaLoadingVisitor();
|
||||
regionStateStore.visitMetaForRegion(regionEncodedName, visitor);
|
||||
}
|
||||
|
||||
private void loadMeta() throws IOException {
|
||||
|
|
@ -1923,11 +1927,23 @@ public class AssignmentManager {
|
|||
return regionStates.getAssignedRegions();
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve a cached {@link RegionInfo} from the region name as a {@code byte[]}.
|
||||
*/
|
||||
public RegionInfo getRegionInfo(final byte[] regionName) {
|
||||
final RegionStateNode regionState = regionStates.getRegionStateNodeFromName(regionName);
|
||||
return regionState != null ? regionState.getRegionInfo() : null;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve a cached {@link RegionInfo} from the encoded region name as a {@code String}.
|
||||
*/
|
||||
public RegionInfo getRegionInfo(final String encodedRegionName) {
|
||||
final RegionStateNode regionState =
|
||||
regionStates.getRegionStateNodeFromEncodedRegionName(encodedRegionName);
|
||||
return regionState != null ? regionState.getRegionInfo() : null;
|
||||
}
|
||||
|
||||
// ============================================================================================
|
||||
// Expected states on region state transition.
|
||||
// Notice that there is expected states for transiting to OPENING state, this is because SCP.
|
||||
|
|
|
|||
|
|
@ -131,9 +131,9 @@ public class RegionStateStore {
|
|||
|
||||
/**
|
||||
* Queries META table for the passed region encoded name, delegating action upon results to the
|
||||
* <code>RegionStateVisitor</code> passed as second parameter.
|
||||
* {@code RegionStateVisitor} passed as second parameter.
|
||||
* @param regionEncodedName encoded name for the Region we want to query META for.
|
||||
* @param visitor The <code>RegionStateVisitor</code> instance to react over the query
|
||||
* @param visitor The {@code RegionStateVisitor} instance to react over the query
|
||||
* results.
|
||||
* @throws IOException If some error occurs while querying META or parsing results.
|
||||
*/
|
||||
|
|
@ -360,8 +360,10 @@ public class RegionStateStore {
|
|||
// default OFFLINE state. If Master gets restarted after this step, start up sequence of
|
||||
// master tries to assign these offline regions. This is followed by re-assignments of the
|
||||
// daughter regions from resumed {@link SplitTableRegionProcedure}
|
||||
MetaTableAccessor.addRegionStateToPut(putA, RegionState.State.CLOSED);
|
||||
MetaTableAccessor.addRegionStateToPut(putB, RegionState.State.CLOSED);
|
||||
MetaTableAccessor.addRegionStateToPut(putA, RegionInfo.DEFAULT_REPLICA_ID,
|
||||
RegionState.State.CLOSED);
|
||||
MetaTableAccessor.addRegionStateToPut(putB, RegionInfo.DEFAULT_REPLICA_ID,
|
||||
RegionState.State.CLOSED);
|
||||
|
||||
// new regions, openSeqNum = 1 is fine.
|
||||
addSequenceNum(putA, 1, splitA.getReplicaId());
|
||||
|
|
@ -405,7 +407,8 @@ public class RegionStateStore {
|
|||
// default OFFLINE state. If Master gets restarted after this step, start up sequence of
|
||||
// master tries to assign this offline region. This is followed by re-assignments of the
|
||||
// merged region from resumed {@link MergeTableRegionsProcedure}
|
||||
MetaTableAccessor.addRegionStateToPut(putOfMerged, RegionState.State.CLOSED);
|
||||
MetaTableAccessor.addRegionStateToPut(putOfMerged, RegionInfo.DEFAULT_REPLICA_ID,
|
||||
RegionState.State.CLOSED);
|
||||
mutations.add(putOfMerged);
|
||||
// The merged is a new region, openSeqNum = 1 is fine. ServerName may be null
|
||||
// if crash after merge happened but before we got to here.. means in-memory
|
||||
|
|
|
|||
|
|
@ -72,7 +72,7 @@ public class RegionStates {
|
|||
|
||||
// TODO: Replace the ConcurrentSkipListMaps
|
||||
/**
|
||||
* RegionName -- i.e. RegionInfo.getRegionName() -- as bytes to {@link RegionStateNode}
|
||||
* A Map from {@link RegionInfo#getRegionName()} to {@link RegionStateNode}
|
||||
*/
|
||||
private final ConcurrentSkipListMap<byte[], RegionStateNode> regionsMap =
|
||||
new ConcurrentSkipListMap<>(Bytes.BYTES_COMPARATOR);
|
||||
|
|
@ -144,6 +144,10 @@ public class RegionStates {
|
|||
return regionsMap.get(regionName);
|
||||
}
|
||||
|
||||
public RegionStateNode getRegionStateNodeFromEncodedRegionName(final String encodedRegionName) {
|
||||
return encodedRegionsMap.get(encodedRegionName);
|
||||
}
|
||||
|
||||
public RegionStateNode getRegionStateNode(RegionInfo regionInfo) {
|
||||
return getRegionStateNodeFromName(regionInfo.getRegionName());
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1918,7 +1918,7 @@ public class HBaseFsck extends Configured implements Closeable {
|
|||
|
||||
RegionInfo hri = RegionInfoBuilder.newBuilder(hi.getMetaEntry().getRegionInfo())
|
||||
.setOffline(false).setSplit(false).build();
|
||||
Put p = MetaTableAccessor.makePutFromRegionInfo(hri, EnvironmentEdgeManager.currentTime());
|
||||
Put p = MetaTableAccessor.makePutFromRegionInfo(hri);
|
||||
mutations.add(p);
|
||||
|
||||
meta.mutateRow(mutations);
|
||||
|
|
|
|||
|
|
@ -150,7 +150,7 @@ public class HBaseFsckRepair {
|
|||
Collection<ServerName> servers, int numReplicas) throws IOException {
|
||||
Connection conn = ConnectionFactory.createConnection(conf);
|
||||
Table meta = conn.getTable(TableName.META_TABLE_NAME);
|
||||
Put put = MetaTableAccessor.makePutFromRegionInfo(hri, EnvironmentEdgeManager.currentTime());
|
||||
Put put = MetaTableAccessor.makePutFromRegionInfo(hri);
|
||||
if (numReplicas > 1) {
|
||||
Random rand = ThreadLocalRandom.current();
|
||||
ServerName[] serversArr = servers.toArray(new ServerName[servers.size()]);
|
||||
|
|
|
|||
|
|
@ -29,6 +29,7 @@ import java.util.HashMap;
|
|||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.CountDownLatch;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import java.util.stream.Collectors;
|
||||
|
|
@ -57,7 +58,6 @@ import org.apache.hadoop.hbase.regionserver.HRegionServer;
|
|||
import org.apache.hadoop.hbase.testclassification.ClientTests;
|
||||
import org.apache.hadoop.hbase.testclassification.LargeTests;
|
||||
import org.apache.hadoop.hbase.util.Bytes;
|
||||
import org.apache.hadoop.hbase.util.Pair;
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.Before;
|
||||
import org.junit.BeforeClass;
|
||||
|
|
@ -117,7 +117,7 @@ public class TestHbck {
|
|||
@BeforeClass
|
||||
public static void setUpBeforeClass() throws Exception {
|
||||
TEST_UTIL.startMiniCluster(3);
|
||||
TEST_UTIL.createMultiRegionTable(TABLE_NAME, Bytes.toBytes("family1"), 5);
|
||||
TEST_UTIL.createMultiRegionTable(TABLE_NAME, 3, new byte[][] { Bytes.toBytes("family1") });
|
||||
procExec = TEST_UTIL.getMiniHBaseCluster().getMaster().getMasterProcedureExecutor();
|
||||
ASYNC_CONN = ConnectionFactory.createAsyncConnection(TEST_UTIL.getConfiguration()).get();
|
||||
TEST_UTIL.getHBaseCluster().getMaster().getMasterCoprocessorHost().load(
|
||||
|
|
@ -187,7 +187,7 @@ public class TestHbck {
|
|||
// will be DISABLED
|
||||
TableState prevState =
|
||||
hbck.setTableStateInMeta(new TableState(TABLE_NAME, TableState.State.ENABLED));
|
||||
assertTrue("Incorrect previous state! expeced=DISABLED, found=" + prevState.getState(),
|
||||
assertTrue("Incorrect previous state! expected=DISABLED, found=" + prevState.getState(),
|
||||
prevState.isDisabled());
|
||||
}
|
||||
|
||||
|
|
@ -195,48 +195,57 @@ public class TestHbck {
|
|||
public void testSetRegionStateInMeta() throws Exception {
|
||||
Hbck hbck = getHbck();
|
||||
Admin admin = TEST_UTIL.getAdmin();
|
||||
TEST_UTIL.waitUntilAllRegionsAssigned(TABLE_NAME);
|
||||
final List<RegionInfo> regions = admin.getRegions(TABLE_NAME);
|
||||
final AssignmentManager am = TEST_UTIL.getHBaseCluster().getMaster().getAssignmentManager();
|
||||
Map<String, RegionState.State> prevStates = new HashMap<>();
|
||||
Map<String, RegionState.State> newStates = new HashMap<>();
|
||||
final Map<String, Pair<RegionState.State, RegionState.State>> regionsMap = new HashMap<>();
|
||||
final Map<String, RegionState.State> beforeStates = new HashMap<>();
|
||||
final Map<String, RegionState.State> requestStates = new HashMap<>();
|
||||
regions.forEach(r -> {
|
||||
RegionState prevState = am.getRegionStates().getRegionState(r);
|
||||
prevStates.put(r.getEncodedName(), prevState.getState());
|
||||
newStates.put(r.getEncodedName(), RegionState.State.CLOSED);
|
||||
regionsMap.put(r.getEncodedName(),
|
||||
new Pair<>(prevState.getState(), RegionState.State.CLOSED));
|
||||
RegionState beforeState = am.getRegionStates().getRegionState(r);
|
||||
beforeStates.put(r.getEncodedName(), beforeState.getState());
|
||||
LOG.debug("Before test: {} ; {}", r, beforeState.getState());
|
||||
requestStates.put(r.getEncodedName(), RegionState.State.CLOSED);
|
||||
});
|
||||
final Map<String, RegionState.State> result = hbck.setRegionStateInMeta(newStates);
|
||||
result.forEach((k, v) -> {
|
||||
RegionState.State prevState = regionsMap.get(k).getFirst();
|
||||
assertEquals(prevState, v);
|
||||
});
|
||||
regions.forEach(r -> {
|
||||
RegionState cachedState = am.getRegionStates().getRegionState(r.getEncodedName());
|
||||
RegionState.State newState = regionsMap.get(r.getEncodedName()).getSecond();
|
||||
assertEquals(newState, cachedState.getState());
|
||||
});
|
||||
hbck.setRegionStateInMeta(prevStates);
|
||||
final Callable<Void> doTest = () -> {
|
||||
// run the entire test with the ProcedureExecution environment paused. This prevents
|
||||
// background operations from modifying AM internal state between the assertions this test
|
||||
// relies upon.
|
||||
Map<String, RegionState.State> result = hbck.setRegionStateInMeta(requestStates);
|
||||
result.forEach((k, v) -> {
|
||||
RegionState.State beforeState = beforeStates.get(k);
|
||||
assertEquals("response state should match before state; " + k, beforeState, v);
|
||||
});
|
||||
regions.forEach(r -> {
|
||||
RegionState afterState = am.getRegionStates().getRegionState(r.getEncodedName());
|
||||
RegionState.State expectedState = requestStates.get(r.getEncodedName());
|
||||
LOG.debug("After test: {}, {}", r, afterState);
|
||||
assertEquals("state in AM should match requested state ; " + r, expectedState,
|
||||
afterState.getState());
|
||||
});
|
||||
return null;
|
||||
};
|
||||
ProcedureTestingUtility.restart(procExec, true, true, null, doTest, null, false, true);
|
||||
// restore the table as we found it -- fragile?
|
||||
hbck.setRegionStateInMeta(beforeStates);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testAssigns() throws Exception {
|
||||
Hbck hbck = getHbck();
|
||||
final AssignmentManager am = TEST_UTIL.getHBaseCluster().getMaster().getAssignmentManager();
|
||||
try (Admin admin = TEST_UTIL.getConnection().getAdmin()) {
|
||||
List<RegionInfo> regions = admin.getRegions(TABLE_NAME);
|
||||
for (RegionInfo ri : regions) {
|
||||
RegionState rs = TEST_UTIL.getHBaseCluster().getMaster().getAssignmentManager()
|
||||
.getRegionStates().getRegionState(ri.getEncodedName());
|
||||
LOG.info("RS: {}", rs.toString());
|
||||
}
|
||||
List<Long> pids =
|
||||
hbck.unassigns(regions.stream().map(r -> r.getEncodedName()).collect(Collectors.toList()));
|
||||
List<RegionInfo> regions = admin.getRegions(TABLE_NAME).stream()
|
||||
.filter(ri -> ri.getReplicaId() == RegionInfo.DEFAULT_REPLICA_ID).peek(ri -> {
|
||||
final RegionState rs = am.getRegionStates().getRegionState(ri.getEncodedName());
|
||||
LOG.info("RS: {}", rs);
|
||||
}).collect(Collectors.toList());
|
||||
List<Long> pids = hbck
|
||||
.unassigns(regions.stream().map(RegionInfo::getEncodedName).collect(Collectors.toList()));
|
||||
waitOnPids(pids);
|
||||
// Rerun the unassign. Should fail for all Regions since they already unassigned; failed
|
||||
// unassign will manifest as all pids being -1 (ever since HBASE-24885).
|
||||
pids =
|
||||
hbck.unassigns(regions.stream().map(r -> r.getEncodedName()).collect(Collectors.toList()));
|
||||
pids = hbck
|
||||
.unassigns(regions.stream().map(RegionInfo::getEncodedName).collect(Collectors.toList()));
|
||||
waitOnPids(pids);
|
||||
for (long pid : pids) {
|
||||
assertEquals(Procedure.NO_PROC_ID, pid);
|
||||
|
|
@ -245,7 +254,7 @@ public class TestHbck {
|
|||
// unassigned.... makes for a mess but operator might want to do this at an extreme when
|
||||
// doing fixup of broke cluster.
|
||||
pids = hbck.unassigns(
|
||||
regions.stream().map(r -> r.getEncodedName()).collect(Collectors.toList()), true);
|
||||
regions.stream().map(RegionInfo::getEncodedName).collect(Collectors.toList()), true);
|
||||
waitOnPids(pids);
|
||||
for (long pid : pids) {
|
||||
assertNotEquals(Procedure.NO_PROC_ID, pid);
|
||||
|
|
@ -259,12 +268,12 @@ public class TestHbck {
|
|||
assertTrue(rs.toString(), rs.isClosed());
|
||||
}
|
||||
pids =
|
||||
hbck.assigns(regions.stream().map(r -> r.getEncodedName()).collect(Collectors.toList()));
|
||||
hbck.assigns(regions.stream().map(RegionInfo::getEncodedName).collect(Collectors.toList()));
|
||||
waitOnPids(pids);
|
||||
// Rerun the assign. Should fail for all Regions since they already assigned; failed
|
||||
// assign will manifest as all pids being -1 (ever since HBASE-24885).
|
||||
pids =
|
||||
hbck.assigns(regions.stream().map(r -> r.getEncodedName()).collect(Collectors.toList()));
|
||||
hbck.assigns(regions.stream().map(RegionInfo::getEncodedName).collect(Collectors.toList()));
|
||||
for (long pid : pids) {
|
||||
assertEquals(Procedure.NO_PROC_ID, pid);
|
||||
}
|
||||
|
|
@ -286,17 +295,19 @@ public class TestHbck {
|
|||
@Test
|
||||
public void testScheduleSCP() throws Exception {
|
||||
HRegionServer testRs = TEST_UTIL.getRSForFirstRegionInTable(TABLE_NAME);
|
||||
TEST_UTIL.loadTable(TEST_UTIL.getConnection().getTable(TABLE_NAME), Bytes.toBytes("family1"),
|
||||
true);
|
||||
try (final Table t = TEST_UTIL.getConnection().getTable(TABLE_NAME)) {
|
||||
TEST_UTIL.loadTable(t, Bytes.toBytes("family1"), true);
|
||||
}
|
||||
ServerName serverName = testRs.getServerName();
|
||||
Hbck hbck = getHbck();
|
||||
List<Long> pids = hbck.scheduleServerCrashProcedures(Arrays.asList(serverName));
|
||||
assertTrue(pids.get(0) > 0);
|
||||
LOG.info("pid is {}", pids.get(0));
|
||||
assertEquals(1, pids.size());
|
||||
assertNotEquals((Long) Procedure.NO_PROC_ID, pids.get(0));
|
||||
LOG.debug("SCP pid is {}", pids.get(0));
|
||||
|
||||
List<Long> newPids = hbck.scheduleServerCrashProcedures(Arrays.asList(serverName));
|
||||
assertTrue(newPids.get(0) < 0);
|
||||
LOG.info("pid is {}", newPids.get(0));
|
||||
assertEquals(1, pids.size());
|
||||
assertEquals((Long) Procedure.NO_PROC_ID, newPids.get(0));
|
||||
waitOnPids(pids);
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -140,7 +140,7 @@ public class TestRegionsRecoveryChore {
|
|||
|
||||
// Verify that we need to reopen total 3 regions that have refCount > 300
|
||||
Mockito.verify(hMaster, Mockito.times(3)).getAssignmentManager();
|
||||
Mockito.verify(assignmentManager, Mockito.times(3)).getRegionInfo(Mockito.any());
|
||||
Mockito.verify(assignmentManager, Mockito.times(3)).getRegionInfo(Mockito.any(byte[].class));
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -175,7 +175,7 @@ public class TestRegionsRecoveryChore {
|
|||
|
||||
// Verify that we need to reopen only 1 region with refCount > 400
|
||||
Mockito.verify(hMaster, Mockito.times(1)).getAssignmentManager();
|
||||
Mockito.verify(assignmentManager, Mockito.times(1)).getRegionInfo(Mockito.any());
|
||||
Mockito.verify(assignmentManager, Mockito.times(1)).getRegionInfo(Mockito.any(byte[].class));
|
||||
}
|
||||
|
||||
@Test
|
||||
|
|
@ -210,7 +210,7 @@ public class TestRegionsRecoveryChore {
|
|||
|
||||
// default maxCompactedStoreFileRefCount is -1 (no regions to be reopened using AM)
|
||||
Mockito.verify(hMaster, Mockito.times(0)).getAssignmentManager();
|
||||
Mockito.verify(assignmentManager, Mockito.times(0)).getRegionInfo(Mockito.any());
|
||||
Mockito.verify(assignmentManager, Mockito.times(0)).getRegionInfo(Mockito.any(byte[].class));
|
||||
}
|
||||
|
||||
private static ClusterMetrics getClusterMetrics(int noOfLiveServer) {
|
||||
|
|
|
|||
|
|
@ -18,6 +18,7 @@
|
|||
package org.apache.hadoop.hbase.master.assignment;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
import static org.junit.Assert.assertNotNull;
|
||||
import static org.junit.Assert.assertNull;
|
||||
import static org.junit.Assert.assertTrue;
|
||||
|
||||
|
|
@ -303,12 +304,13 @@ public class TestAssignmentManager extends TestAssignmentManagerBase {
|
|||
+ "shouldn't be in AM regionStates yet.", am.getRegionStates().getRegionState(hri));
|
||||
MetaTableAccessor.addRegionsToMeta(this.util.getConnection(), Collections.singletonList(hri),
|
||||
1);
|
||||
assertNull(
|
||||
"RegionInfo was manually added in META, but " + "shouldn't be in AM regionStates yet.",
|
||||
// TODO: is there a race here -- no other thread else will refresh the table states behind
|
||||
// the scenes?
|
||||
assertNull("RegionInfo was manually added in META, but shouldn't be in AM regionStates yet.",
|
||||
am.getRegionStates().getRegionState(hri));
|
||||
hri = am.loadRegionFromMeta(hri.getEncodedName());
|
||||
assertEquals(hri.getEncodedName(),
|
||||
am.getRegionStates().getRegionState(hri).getRegion().getEncodedName());
|
||||
am.populateRegionStatesFromMeta(hri.getEncodedName());
|
||||
assertNotNull(am.getRegionInfo(hri.getRegionName()));
|
||||
assertNotNull(am.getRegionInfo(hri.getEncodedName()));
|
||||
} finally {
|
||||
this.util.killMiniHBaseCluster();
|
||||
}
|
||||
|
|
@ -322,10 +324,10 @@ public class TestAssignmentManager extends TestAssignmentManagerBase {
|
|||
final TableName tableName = TableName.valueOf("testLoadRegionFromMetaRegionNotInMeta");
|
||||
this.util.createTable(tableName, "f");
|
||||
final RegionInfo hri = createRegionInfo(tableName, 1);
|
||||
assertNull("RegionInfo was just instantiated by the test, but "
|
||||
+ "shouldn't be in AM regionStates yet.", am.getRegionStates().getRegionState(hri));
|
||||
assertNull("RegionInfo was never added in META, should had returned null.",
|
||||
am.loadRegionFromMeta(hri.getEncodedName()));
|
||||
assertNull("Bogus RegionInfo discovered in RegionStates.",
|
||||
am.getRegionStates().getRegionState(hri));
|
||||
am.populateRegionStatesFromMeta(hri.getEncodedName());
|
||||
assertNull("RegionInfo was never added in META", am.getRegionStates().getRegionState(hri));
|
||||
} finally {
|
||||
this.util.killMiniHBaseCluster();
|
||||
}
|
||||
|
|
|
|||
|
|
@ -25,31 +25,38 @@ import static org.junit.Assert.assertTrue;
|
|||
|
||||
import java.io.IOException;
|
||||
import java.util.List;
|
||||
import java.util.NoSuchElementException;
|
||||
import java.util.Objects;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
import org.apache.hadoop.hbase.CatalogFamilyFormat;
|
||||
import org.apache.hadoop.hbase.HBaseClassTestRule;
|
||||
import org.apache.hadoop.hbase.HBaseServerBase;
|
||||
import org.apache.hadoop.hbase.HBaseTestingUtil;
|
||||
import org.apache.hadoop.hbase.HConstants;
|
||||
import org.apache.hadoop.hbase.MetaTableAccessor;
|
||||
import org.apache.hadoop.hbase.ServerName;
|
||||
import org.apache.hadoop.hbase.SingleProcessHBaseCluster;
|
||||
import org.apache.hadoop.hbase.TableName;
|
||||
import org.apache.hadoop.hbase.TableNameTestRule;
|
||||
import org.apache.hadoop.hbase.client.RegionInfo;
|
||||
import org.apache.hadoop.hbase.client.Result;
|
||||
import org.apache.hadoop.hbase.client.Table;
|
||||
import org.apache.hadoop.hbase.master.HMaster;
|
||||
import org.apache.hadoop.hbase.master.RegionState;
|
||||
import org.apache.hadoop.hbase.procedure2.Procedure;
|
||||
import org.apache.hadoop.hbase.procedure2.ProcedureTestingUtility;
|
||||
import org.apache.hadoop.hbase.regionserver.HRegionServer;
|
||||
import org.apache.hadoop.hbase.testclassification.LargeTests;
|
||||
import org.apache.hadoop.hbase.testclassification.MasterTests;
|
||||
import org.apache.hadoop.hbase.util.Bytes;
|
||||
import org.apache.hadoop.hbase.util.JVMClusterUtil;
|
||||
import org.apache.hadoop.hbase.util.Pair;
|
||||
import org.apache.hadoop.hbase.util.Threads;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.experimental.categories.Category;
|
||||
import org.junit.rules.TestName;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.junit.runners.Parameterized;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
|
||||
|
|
@ -63,6 +70,7 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos;
|
|||
* Regions that were on the server-to-process rather than consult Master in-memory-state.
|
||||
*/
|
||||
@Category({ MasterTests.class, LargeTests.class })
|
||||
@RunWith(Parameterized.class)
|
||||
public class TestHBCKSCP extends TestSCPBase {
|
||||
private static final Logger LOG = LoggerFactory.getLogger(TestHBCKSCP.class);
|
||||
|
||||
|
|
@ -70,7 +78,27 @@ public class TestHBCKSCP extends TestSCPBase {
|
|||
public static final HBaseClassTestRule CLASS_RULE =
|
||||
HBaseClassTestRule.forClass(TestHBCKSCP.class);
|
||||
@Rule
|
||||
public TestName name = new TestName();
|
||||
public TableNameTestRule tableNameTestRule = new TableNameTestRule();
|
||||
|
||||
private final int replicas;
|
||||
private final HBCKSCPScheduler hbckscpScheduler;
|
||||
private final RegionSelector regionSelector;
|
||||
|
||||
public TestHBCKSCP(final int replicas, final HBCKSCPScheduler hbckscpScheduler,
|
||||
final RegionSelector regionSelector) {
|
||||
this.replicas = replicas;
|
||||
this.hbckscpScheduler = hbckscpScheduler;
|
||||
this.regionSelector = regionSelector;
|
||||
}
|
||||
|
||||
@Parameterized.Parameters(name = "replicas:{0} scheduler:{1} selector:{2}")
|
||||
public static Object[][] params() {
|
||||
return new Object[][] {
|
||||
{ 1, new ScheduleServerCrashProcedure(), new PrimaryNotMetaRegionSelector() },
|
||||
{ 3, new ScheduleServerCrashProcedure(), new ReplicaNonMetaRegionSelector() },
|
||||
{ 1, new ScheduleSCPsForUnknownServers(), new PrimaryNotMetaRegionSelector() },
|
||||
{ 3, new ScheduleSCPsForUnknownServers(), new ReplicaNonMetaRegionSelector() } };
|
||||
}
|
||||
|
||||
@Test
|
||||
public void test() throws Exception {
|
||||
|
|
@ -81,10 +109,10 @@ public class TestHBCKSCP extends TestSCPBase {
|
|||
assertEquals(RS_COUNT, cluster.getLiveRegionServerThreads().size());
|
||||
|
||||
int count;
|
||||
try (Table table = createTable(TableName.valueOf(this.name.getMethodName()))) {
|
||||
try (Table table = createTable(tableNameTestRule.getTableName())) {
|
||||
// Load the table with a bit of data so some logs to split and some edits in each region.
|
||||
this.util.loadTable(table, HBaseTestingUtil.COLUMNS[0]);
|
||||
count = util.countRows(table);
|
||||
count = HBaseTestingUtil.countRows(table);
|
||||
}
|
||||
assertTrue("expected some rows", count > 0);
|
||||
|
||||
|
|
@ -92,17 +120,24 @@ public class TestHBCKSCP extends TestSCPBase {
|
|||
// Find another RS. Purge it from Master memory w/o running SCP (if
|
||||
// SCP runs, it will clear entries from hbase:meta which frustrates
|
||||
// our attempt at manufacturing 'Unknown Servers' condition).
|
||||
int metaIndex = this.util.getMiniHBaseCluster().getServerWithMeta();
|
||||
int rsIndex = (metaIndex + 1) % RS_COUNT;
|
||||
ServerName rsServerName = cluster.getRegionServer(rsIndex).getServerName();
|
||||
final ServerName metaServer = util.getMiniHBaseCluster().getServerHoldingMeta();
|
||||
final ServerName rsServerName = cluster.getRegionServerThreads().stream()
|
||||
.map(JVMClusterUtil.RegionServerThread::getRegionServer).map(HBaseServerBase::getServerName)
|
||||
.filter(sn -> !sn.equals(metaServer)).findAny().orElseThrow(() -> new NoSuchElementException(
|
||||
"Cannot locate a region server that is not hosting meta."));
|
||||
HMaster master = cluster.getMaster();
|
||||
// Get a Region that is on the server.
|
||||
RegionInfo rsRI = master.getAssignmentManager().getRegionsOnServer(rsServerName).get(0);
|
||||
Result r = MetaTableAccessor.getRegionResult(master.getConnection(), rsRI.getRegionName());
|
||||
final List<RegionInfo> regions = master.getAssignmentManager().getRegionsOnServer(rsServerName);
|
||||
LOG.debug("{} is holding {} regions.", rsServerName, regions.size());
|
||||
final RegionInfo rsRI =
|
||||
regions.stream().peek(info -> LOG.debug("{}", info)).filter(regionSelector::regionFilter)
|
||||
.findAny().orElseThrow(regionSelector::regionFilterFailure);
|
||||
final int replicaId = rsRI.getReplicaId();
|
||||
Result r = MetaTableAccessor.getRegionResult(master.getConnection(), rsRI);
|
||||
// Assert region is OPEN.
|
||||
assertEquals(RegionState.State.OPEN.toString(),
|
||||
Bytes.toString(r.getValue(HConstants.CATALOG_FAMILY, HConstants.STATE_QUALIFIER)));
|
||||
ServerName serverName = CatalogFamilyFormat.getServerName(r, 0);
|
||||
assertEquals(RegionState.State.OPEN.toString(), Bytes.toString(
|
||||
r.getValue(HConstants.CATALOG_FAMILY, CatalogFamilyFormat.getRegionStateColumn(replicaId))));
|
||||
ServerName serverName = CatalogFamilyFormat.getServerName(r, replicaId);
|
||||
assertEquals(rsServerName, serverName);
|
||||
// moveFrom adds to dead servers and adds it to processing list only we will
|
||||
// not be processing this server 'normally'. Remove it from processing by
|
||||
|
|
@ -117,18 +152,16 @@ public class TestHBCKSCP extends TestSCPBase {
|
|||
// Kill the server. Nothing should happen since an 'Unknown Server' as far
|
||||
// as the Master is concerned; i.e. no SCP.
|
||||
HRegionServer hrs = cluster.getRegionServer(rsServerName);
|
||||
while (!hrs.isStopped()) {
|
||||
Threads.sleep(10);
|
||||
}
|
||||
util.waitFor(TimeUnit.MINUTES.toMillis(1), hrs::isStopped);
|
||||
LOG.info("Dead {}", rsServerName);
|
||||
// Now assert still references in hbase:meta to the 'dead' server -- they haven't been
|
||||
// cleaned up by an SCP or by anything else.
|
||||
assertTrue(searchMeta(master, rsServerName));
|
||||
// Assert region is OPEN on dead server still.
|
||||
r = MetaTableAccessor.getRegionResult(master.getConnection(), rsRI.getRegionName());
|
||||
assertEquals(RegionState.State.OPEN.toString(),
|
||||
Bytes.toString(r.getValue(HConstants.CATALOG_FAMILY, HConstants.STATE_QUALIFIER)));
|
||||
serverName = CatalogFamilyFormat.getServerName(r, 0);
|
||||
r = MetaTableAccessor.getRegionResult(master.getConnection(), rsRI);
|
||||
assertEquals(RegionState.State.OPEN.toString(), Bytes.toString(
|
||||
r.getValue(HConstants.CATALOG_FAMILY, CatalogFamilyFormat.getRegionStateColumn(replicaId))));
|
||||
serverName = CatalogFamilyFormat.getServerName(r, replicaId);
|
||||
assertNotNull(cluster.getRegionServer(serverName));
|
||||
assertEquals(rsServerName, serverName);
|
||||
|
||||
|
|
@ -136,13 +169,11 @@ public class TestHBCKSCP extends TestSCPBase {
|
|||
// with no corresponding SCP. Queue one.
|
||||
long pid = scheduleHBCKSCP(rsServerName, master);
|
||||
assertNotEquals(Procedure.NO_PROC_ID, pid);
|
||||
while (master.getMasterProcedureExecutor().getActiveProcIds().contains(pid)) {
|
||||
Threads.sleep(10);
|
||||
}
|
||||
ProcedureTestingUtility.waitProcedure(master.getMasterProcedureExecutor(), pid);
|
||||
// After SCP, assert region is OPEN on new server.
|
||||
r = MetaTableAccessor.getRegionResult(master.getConnection(), rsRI.getRegionName());
|
||||
assertEquals(RegionState.State.OPEN.toString(),
|
||||
Bytes.toString(r.getValue(HConstants.CATALOG_FAMILY, HConstants.STATE_QUALIFIER)));
|
||||
r = MetaTableAccessor.getRegionResult(master.getConnection(), rsRI);
|
||||
assertEquals(RegionState.State.OPEN.toString(), Bytes.toString(
|
||||
r.getValue(HConstants.CATALOG_FAMILY, CatalogFamilyFormat.getRegionStateColumn(replicaId))));
|
||||
serverName = CatalogFamilyFormat.getServerName(r, 0);
|
||||
assertNotNull(cluster.getRegionServer(serverName));
|
||||
assertNotEquals(rsServerName, serverName);
|
||||
|
|
@ -151,12 +182,12 @@ public class TestHBCKSCP extends TestSCPBase {
|
|||
}
|
||||
|
||||
protected long scheduleHBCKSCP(ServerName rsServerName, HMaster master) throws ServiceException {
|
||||
MasterProtos.ScheduleServerCrashProcedureResponse response = master.getMasterRpcServices()
|
||||
.scheduleServerCrashProcedure(null, MasterProtos.ScheduleServerCrashProcedureRequest
|
||||
.newBuilder().addServerName(ProtobufUtil.toServerName(rsServerName)).build());
|
||||
assertEquals(1, response.getPidCount());
|
||||
long pid = response.getPid(0);
|
||||
return pid;
|
||||
return hbckscpScheduler.scheduleHBCKSCP(rsServerName, master);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected int getRegionReplication() {
|
||||
return replicas;
|
||||
}
|
||||
|
||||
/** Returns True if we find reference to <code>sn</code> in meta table. */
|
||||
|
|
@ -170,4 +201,90 @@ public class TestHBCKSCP extends TestSCPBase {
|
|||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/**
|
||||
* Encapsulates the choice of which HBCK2 method to call.
|
||||
*/
|
||||
private abstract static class HBCKSCPScheduler {
|
||||
abstract long scheduleHBCKSCP(ServerName rsServerName, HMaster master) throws ServiceException;
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return this.getClass().getSimpleName();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Invokes {@code MasterRpcServices#scheduleServerCrashProcedure}.
|
||||
*/
|
||||
private static class ScheduleServerCrashProcedure extends HBCKSCPScheduler {
|
||||
@Override
|
||||
public long scheduleHBCKSCP(ServerName rsServerName, HMaster master) throws ServiceException {
|
||||
MasterProtos.ScheduleServerCrashProcedureResponse response = master.getMasterRpcServices()
|
||||
.scheduleServerCrashProcedure(null, MasterProtos.ScheduleServerCrashProcedureRequest
|
||||
.newBuilder().addServerName(ProtobufUtil.toServerName(rsServerName)).build());
|
||||
assertEquals(1, response.getPidCount());
|
||||
return response.getPid(0);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Invokes {@code MasterRpcServices#scheduleSCPsForUnknownServers}.
|
||||
*/
|
||||
private static class ScheduleSCPsForUnknownServers extends HBCKSCPScheduler {
|
||||
@Override
|
||||
long scheduleHBCKSCP(ServerName rsServerName, HMaster master) throws ServiceException {
|
||||
MasterProtos.ScheduleSCPsForUnknownServersResponse response =
|
||||
master.getMasterRpcServices().scheduleSCPsForUnknownServers(null,
|
||||
MasterProtos.ScheduleSCPsForUnknownServersRequest.newBuilder().build());
|
||||
assertEquals(1, response.getPidCount());
|
||||
return response.getPid(0);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Encapsulates how the target region is selected.
|
||||
*/
|
||||
private static abstract class RegionSelector {
|
||||
abstract boolean regionFilter(RegionInfo info);
|
||||
|
||||
abstract Exception regionFilterFailure();
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return this.getClass().getSimpleName();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Selects a non-meta region that is also a primary region.
|
||||
*/
|
||||
private static class PrimaryNotMetaRegionSelector extends RegionSelector {
|
||||
@Override
|
||||
boolean regionFilter(final RegionInfo info) {
|
||||
return !Objects.equals(TableName.META_TABLE_NAME, info.getTable())
|
||||
&& Objects.equals(RegionInfo.DEFAULT_REPLICA_ID, info.getReplicaId());
|
||||
}
|
||||
|
||||
@Override
|
||||
Exception regionFilterFailure() {
|
||||
return new NoSuchElementException("Cannot locate a primary, non-meta region.");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Selects a non-meta region that is also a replica region.
|
||||
*/
|
||||
private static class ReplicaNonMetaRegionSelector extends RegionSelector {
|
||||
@Override
|
||||
boolean regionFilter(RegionInfo info) {
|
||||
return !Objects.equals(TableName.META_TABLE_NAME, info.getTable())
|
||||
&& !Objects.equals(RegionInfo.DEFAULT_REPLICA_ID, info.getReplicaId());
|
||||
}
|
||||
|
||||
@Override
|
||||
Exception regionFilterFailure() {
|
||||
return new NoSuchElementException("Cannot locate a replica, non-meta region.");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,54 +0,0 @@
|
|||
/*
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
package org.apache.hadoop.hbase.master.procedure;
|
||||
|
||||
import static org.junit.Assert.assertEquals;
|
||||
|
||||
import org.apache.hadoop.hbase.HBaseClassTestRule;
|
||||
import org.apache.hadoop.hbase.ServerName;
|
||||
import org.apache.hadoop.hbase.master.HMaster;
|
||||
import org.apache.hadoop.hbase.testclassification.LargeTests;
|
||||
import org.apache.hadoop.hbase.testclassification.MasterTests;
|
||||
import org.junit.ClassRule;
|
||||
import org.junit.experimental.categories.Category;
|
||||
|
||||
import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
|
||||
|
||||
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos;
|
||||
|
||||
/**
|
||||
* Test of the HBCK-version of SCP. The HBCKSCP is an SCP only it reads hbase:meta for list of
|
||||
* Regions that were on the server-to-process rather than consult Master in-memory-state.
|
||||
*/
|
||||
@Category({ MasterTests.class, LargeTests.class })
|
||||
public class TestHBCKSCPUnknown extends TestHBCKSCP {
|
||||
|
||||
@ClassRule
|
||||
public static final HBaseClassTestRule CLASS_RULE =
|
||||
HBaseClassTestRule.forClass(TestHBCKSCPUnknown.class);
|
||||
|
||||
@Override
|
||||
protected long scheduleHBCKSCP(ServerName rsServerName, HMaster master) throws ServiceException {
|
||||
MasterProtos.ScheduleSCPsForUnknownServersResponse response =
|
||||
master.getMasterRpcServices().scheduleSCPsForUnknownServers(null,
|
||||
MasterProtos.ScheduleSCPsForUnknownServersRequest.newBuilder().build());
|
||||
assertEquals(1, response.getPidCount());
|
||||
long pid = response.getPid(0);
|
||||
return pid;
|
||||
}
|
||||
}
|
||||
|
|
@ -224,7 +224,7 @@ public class TestRegionMergeTransactionOnCluster {
|
|||
RegionInfo mergedRegionInfo = tableRegions.get(0).getFirst();
|
||||
TableDescriptor tableDescriptor = MASTER.getTableDescriptors().get(tableName);
|
||||
Result mergedRegionResult =
|
||||
MetaTableAccessor.getRegionResult(MASTER.getConnection(), mergedRegionInfo.getRegionName());
|
||||
MetaTableAccessor.getRegionResult(MASTER.getConnection(), mergedRegionInfo);
|
||||
|
||||
// contains merge reference in META
|
||||
assertTrue(CatalogFamilyFormat.hasMergeRegions(mergedRegionResult.rawCells()));
|
||||
|
|
@ -299,8 +299,8 @@ public class TestRegionMergeTransactionOnCluster {
|
|||
|
||||
// Wait around a bit to give stuff a chance to complete.
|
||||
while (true) {
|
||||
mergedRegionResult = MetaTableAccessor.getRegionResult(TEST_UTIL.getConnection(),
|
||||
mergedRegionInfo.getRegionName());
|
||||
mergedRegionResult =
|
||||
MetaTableAccessor.getRegionResult(TEST_UTIL.getConnection(), mergedRegionInfo);
|
||||
if (CatalogFamilyFormat.hasMergeRegions(mergedRegionResult.rawCells())) {
|
||||
LOG.info("Waiting on cleanup of merge columns {}",
|
||||
Arrays.asList(mergedRegionResult.rawCells()).stream().map(c -> c.toString())
|
||||
|
|
|
|||
Loading…
Reference in New Issue