Skip to content

Commit

Permalink
HBASE-28582 ModifyTableProcedure should not reset TRSP on region node…
Browse files Browse the repository at this point in the history
… when closing unused region replicas (#5903)

Signed-off-by: Viraj Jasani <vjasani@apache.org>
(cherry picked from commit c4a7606)
  • Loading branch information
Apache9 committed May 29, 2024
1 parent ad88fca commit f9af303
Show file tree
Hide file tree
Showing 6 changed files with 327 additions and 10 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -780,3 +780,13 @@ enum MigrateNamespaceTableProcedureState {

message MigrateNamespaceTableProcedureStateData {
}

enum CloseExcessRegionReplicasProcedureState {
CLOSE_EXCESS_REGION_REPLICAS_SCHEDULE = 1;
CLOSE_EXCESS_REGION_REPLICAS_CONFIRM = 2;
}

message CloseExcessRegionReplicasProcedureStateData {
required TableName table_name = 1;
required uint32 new_replica_count = 2;
}
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import org.apache.hadoop.conf.Configuration;
Expand Down Expand Up @@ -1076,14 +1077,55 @@ public TransitRegionStateProcedure[] createUnassignProceduresForDisabling(TableN
}

/**
* Called by ModifyTableProcedures to unassign all the excess region replicas for a table.
* Called by ModifyTableProcedure to unassign all the excess region replicas for a table. Will
* skip submit unassign procedure if the region is in transition, so you may need to call this
* method multiple times.
* @param tableName the table for closing excess region replicas
* @param newReplicaCount the new replica count, should be less than current replica count
* @param submit for submitting procedure
* @return the number of regions in transition that we can not schedule unassign procedures
*/
public TransitRegionStateProcedure[] createUnassignProceduresForClosingExcessRegionReplicas(
TableName tableName, int newReplicaCount) {
return regionStates.getTableRegionStateNodes(tableName).stream()
.filter(regionNode -> regionNode.getRegionInfo().getReplicaId() >= newReplicaCount)
.map(this::forceCreateUnssignProcedure).filter(p -> p != null)
.toArray(TransitRegionStateProcedure[]::new);
public int submitUnassignProcedureForClosingExcessRegionReplicas(TableName tableName,
int newReplicaCount, Consumer<TransitRegionStateProcedure> submit) {
int inTransitionCount = 0;
for (RegionStateNode regionNode : regionStates.getTableRegionStateNodes(tableName)) {
regionNode.lock();
try {
if (regionNode.getRegionInfo().getReplicaId() >= newReplicaCount) {
if (regionNode.isInTransition()) {
LOG.debug("skip scheduling unassign procedure for {} when closing excess region "
+ "replicas since it is in transition", regionNode);
inTransitionCount++;
continue;
}
if (regionNode.isInState(State.OFFLINE, State.CLOSED, State.SPLIT)) {
continue;
}
submit.accept(regionNode.setProcedure(TransitRegionStateProcedure
.unassign(getProcedureEnvironment(), regionNode.getRegionInfo())));
}
} finally {
regionNode.unlock();
}
}
return inTransitionCount;
}

public int numberOfUnclosedExcessRegionReplicas(TableName tableName, int newReplicaCount) {
int unclosed = 0;
for (RegionStateNode regionNode : regionStates.getTableRegionStateNodes(tableName)) {
regionNode.lock();
try {
if (regionNode.getRegionInfo().getReplicaId() >= newReplicaCount) {
if (!regionNode.isInState(State.OFFLINE, State.CLOSED, State.SPLIT)) {
unclosed++;
}
}
} finally {
regionNode.unlock();
}
}
return unclosed;
}

public SplitTableRegionProcedure createSplitProcedure(final RegionInfo regionToSplit,
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,159 @@
/*
* 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 java.io.IOException;
import org.apache.commons.lang3.mutable.MutableBoolean;
import org.apache.hadoop.hbase.TableName;
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.util.RetryCounter;
import org.apache.yetus.audience.InterfaceAudience;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProcedureProtos.CloseExcessRegionReplicasProcedureState;
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProcedureProtos.CloseExcessRegionReplicasProcedureStateData;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ProcedureProtos;

/**
* Procedure for close excess region replicas.
*/
@InterfaceAudience.Private
public class CloseExcessRegionReplicasProcedure
extends AbstractStateMachineTableProcedure<CloseExcessRegionReplicasProcedureState> {

private static final Logger LOG =
LoggerFactory.getLogger(CloseExcessRegionReplicasProcedure.class);

private TableName tableName;
private int newReplicaCount;

private RetryCounter retryCounter;

public CloseExcessRegionReplicasProcedure() {
}

public CloseExcessRegionReplicasProcedure(TableName tableName, int newReplicaCount) {
this.tableName = tableName;
this.newReplicaCount = newReplicaCount;
}

@Override
public TableName getTableName() {
return tableName;
}

@Override
public TableOperationType getTableOperationType() {
return TableOperationType.REGION_EDIT;
}

@Override
protected Flow executeFromState(MasterProcedureEnv env,
CloseExcessRegionReplicasProcedureState state)
throws ProcedureSuspendedException, ProcedureYieldException, InterruptedException {
LOG.trace("{} execute state={}", this, state);
switch (state) {
case CLOSE_EXCESS_REGION_REPLICAS_SCHEDULE:
MutableBoolean submitted = new MutableBoolean(false);
int inTransitionCount = env.getAssignmentManager()
.submitUnassignProcedureForClosingExcessRegionReplicas(tableName, newReplicaCount, p -> {
submitted.setTrue();
addChildProcedure(p);
});
if (inTransitionCount > 0 && submitted.isFalse()) {
// we haven't scheduled any unassign procedures and there are still regions in
// transition, sleep for a while and try again
if (retryCounter == null) {
retryCounter = ProcedureUtil.createRetryCounter(env.getMasterConfiguration());
}
long backoffMillis = retryCounter.getBackoffTimeAndIncrementAttempts();
LOG.info(
"There are still {} region(s) in transition for table {} when closing excess"
+ " region replicas, suspend {}secs and try again later",
inTransitionCount, tableName, backoffMillis / 1000);
suspend((int) backoffMillis, true);
}
setNextState(CloseExcessRegionReplicasProcedureState.CLOSE_EXCESS_REGION_REPLICAS_CONFIRM);
return Flow.HAS_MORE_STATE;
case CLOSE_EXCESS_REGION_REPLICAS_CONFIRM:
int unclosedCount = env.getAssignmentManager()
.numberOfUnclosedExcessRegionReplicas(tableName, newReplicaCount);
if (unclosedCount > 0) {
LOG.info("There are still {} unclosed region(s) for table {} when closing excess"
+ " region replicas, continue...");
setNextState(
CloseExcessRegionReplicasProcedureState.CLOSE_EXCESS_REGION_REPLICAS_SCHEDULE);
return Flow.HAS_MORE_STATE;
} else {
return Flow.NO_MORE_STATE;
}
default:
throw new UnsupportedOperationException("unhandled state=" + state);
}
}

@Override
protected synchronized boolean setTimeoutFailure(MasterProcedureEnv env) {
setState(ProcedureProtos.ProcedureState.RUNNABLE);
env.getProcedureScheduler().addFront(this);
return false;
}

@Override
protected void rollbackState(MasterProcedureEnv env,
CloseExcessRegionReplicasProcedureState state) throws IOException, InterruptedException {
throw new UnsupportedOperationException();
}

@Override
protected CloseExcessRegionReplicasProcedureState getState(int stateId) {
return CloseExcessRegionReplicasProcedureState.forNumber(stateId);
}

@Override
protected int getStateId(CloseExcessRegionReplicasProcedureState state) {
return state.getNumber();
}

@Override
protected CloseExcessRegionReplicasProcedureState getInitialState() {
return CloseExcessRegionReplicasProcedureState.CLOSE_EXCESS_REGION_REPLICAS_SCHEDULE;
}

@Override
protected void serializeStateData(ProcedureStateSerializer serializer) throws IOException {
CloseExcessRegionReplicasProcedureStateData data = CloseExcessRegionReplicasProcedureStateData
.newBuilder().setTableName(ProtobufUtil.toProtoTableName(tableName))
.setNewReplicaCount(newReplicaCount).build();
serializer.serialize(data);
}

@Override
protected void deserializeStateData(ProcedureStateSerializer serializer) throws IOException {
CloseExcessRegionReplicasProcedureStateData data =
serializer.deserialize(CloseExcessRegionReplicasProcedureStateData.class);
tableName = ProtobufUtil.toTableName(data.getTableName());
newReplicaCount = data.getNewReplicaCount();
}

}
Original file line number Diff line number Diff line change
Expand Up @@ -525,8 +525,7 @@ private void closeExcessReplicasIfNeeded(MasterProcedureEnv env) {
if (newReplicaCount >= oldReplicaCount) {
return;
}
addChildProcedure(env.getAssignmentManager()
.createUnassignProceduresForClosingExcessRegionReplicas(getTableName(), newReplicaCount));
addChildProcedure(new CloseExcessRegionReplicasProcedure(getTableName(), newReplicaCount));
}

/**
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,107 @@
/*
* 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.assignment;

import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

import java.util.concurrent.CompletableFuture;
import org.apache.hadoop.hbase.HBaseClassTestRule;
import org.apache.hadoop.hbase.HBaseTestingUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.master.RegionState;
import org.apache.hadoop.hbase.master.procedure.CloseExcessRegionReplicasProcedure;
import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
import org.apache.hadoop.hbase.testclassification.MasterTests;
import org.apache.hadoop.hbase.testclassification.MediumTests;
import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.experimental.categories.Category;

/**
* A test to make sure that we will wait for RIT to finish while closing excess region replicas. See
* HBASE-28582 and related issues for more details.
*/
@Category({ MasterTests.class, MediumTests.class })
public class TestReduceExcessRegionReplicasBlockedByRIT {

@ClassRule
public static final HBaseClassTestRule CLASS_RULE =
HBaseClassTestRule.forClass(TestReduceExcessRegionReplicasBlockedByRIT.class);

private static final HBaseTestingUtil UTIL = new HBaseTestingUtil();

private static TableDescriptor TD =
TableDescriptorBuilder.newBuilder(TableName.valueOf("CloseExcessRegionReplicas"))
.setColumnFamily(ColumnFamilyDescriptorBuilder.of("family")).setRegionReplication(4).build();

@BeforeClass
public static void setUp() throws Exception {
UTIL.startMiniCluster(1);
UTIL.getAdmin().createTable(TD);
UTIL.waitTableAvailable(TD.getTableName());
UTIL.waitUntilNoRegionsInTransition();
}

@AfterClass
public static void tearDown() throws Exception {
UTIL.shutdownMiniCluster();
}

@Test
public void testRIT() throws Exception {
RegionStateNode rsn = UTIL.getMiniHBaseCluster().getMaster().getAssignmentManager()
.getRegionStates().getTableRegionStateNodes(TD.getTableName()).stream()
.filter(rn -> rn.getRegionInfo().getReplicaId() > 1).findAny().get();
// fake a TRSP to block the CloseExcessRegionReplicasProcedure
TransitRegionStateProcedure trsp = new TransitRegionStateProcedure();
rsn.setProcedure(trsp);
TableDescriptor newTd = TableDescriptorBuilder.newBuilder(TD).setRegionReplication(2).build();
CompletableFuture<Void> future = UTIL.getAsyncConnection().getAdmin().modifyTable(newTd);
ProcedureExecutor<MasterProcedureEnv> procExec =
UTIL.getMiniHBaseCluster().getMaster().getMasterProcedureExecutor();
UTIL.waitFor(5000, () -> procExec.getProcedures().stream()
.anyMatch(p -> p instanceof CloseExcessRegionReplicasProcedure && !p.isFinished()));
CloseExcessRegionReplicasProcedure proc =
procExec.getProcedures().stream().filter(p -> p instanceof CloseExcessRegionReplicasProcedure)
.map(p -> (CloseExcessRegionReplicasProcedure) p).findFirst().get();
// make sure that the procedure can not finish
for (int i = 0; i < 5; i++) {
Thread.sleep(3000);
assertFalse(proc.isFinished());
}
assertTrue(rsn.isInState(RegionState.State.OPEN));
// unset the procedure, so we could make progress on CloseExcessRegionReplicasProcedure
rsn.unsetProcedure(trsp);
UTIL.waitFor(60000, () -> proc.isFinished());

future.get();

// the region should be in CLOSED state, and should have been removed from AM
assertTrue(rsn.isInState(RegionState.State.CLOSED));
// only 2 replicas now
assertEquals(2, UTIL.getMiniHBaseCluster().getRegions(TD.getTableName()).size());
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -157,7 +157,7 @@ public void testRegionReplicasByEnableTableWhenReplicaCountIsDecreasedWithMultip
}

@Test
public void testRegionReplicasByEnableTableWhenReplicaCountIsIncreasedWithmultipleRegions()
public void testRegionReplicasByEnableTableWhenReplicaCountIsIncreasedWithMultipleRegions()
throws Exception {
enableReplicationByModification(true, 2, 3, 15);
}
Expand Down

0 comments on commit f9af303

Please sign in to comment.