Skip to content

Commit 1db9a2b

Browse files
Apache9Apache9
committed
HBASE-22662 Move RSGroupInfoManager to hbase-server (#368)
Signed-off-by: Guanghao Zhang <[email protected]>
1 parent f21c3f4 commit 1db9a2b

File tree

6 files changed

+71
-83
lines changed

6 files changed

+71
-83
lines changed

‎hbase-rsgroup/pom.xml‎

Lines changed: 0 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -112,10 +112,6 @@
112112
<groupId>org.apache.hbase.thirdparty</groupId>
113113
<artifactId>hbase-shaded-miscellaneous</artifactId>
114114
</dependency>
115-
<dependency>
116-
<groupId>com.google.protobuf</groupId>
117-
<artifactId>protobuf-java</artifactId>
118-
</dependency>
119115
<dependency>
120116
<groupId>org.apache.zookeeper</groupId>
121117
<artifactId>zookeeper</artifactId>

‎hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupBasedLoadBalancer.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -121,8 +121,8 @@ public List<RegionPlan> balanceCluster(TableName tableName, Map<ServerName, List
121121
public List<RegionPlan> balanceCluster(Map<ServerName, List<RegionInfo>> clusterState)
122122
throws HBaseIOException {
123123
if (!isOnline()) {
124-
throw new ConstraintException(RSGroupInfoManager.RSGROUP_TABLE_NAME +
125-
" is not online, unable to perform balance");
124+
throw new ConstraintException(
125+
RSGroupInfoManager.class.getSimpleName() + " is not online, unable to perform balance");
126126
}
127127

128128
// Calculate correct assignments and a list of RegionPlan for mis-placed regions

‎hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/TestRSGroupsOfflineMode.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -140,7 +140,7 @@ public boolean evaluate() throws Exception {
140140
}
141141
});
142142
// Move table to group and wait.
143-
groupAdmin.moveTables(Sets.newHashSet(RSGroupInfoManager.RSGROUP_TABLE_NAME), newGroup);
143+
groupAdmin.moveTables(Sets.newHashSet(RSGroupInfoManagerImpl.RSGROUP_TABLE_NAME), newGroup);
144144
LOG.info("Waiting for move table...");
145145
TEST_UTIL.waitFor(WAIT_TIMEOUT, new Waiter.Predicate<Exception>() {
146146
@Override
@@ -169,7 +169,7 @@ public boolean evaluate() throws Exception {
169169
// Make sure balancer is in offline mode, since this is what we're testing.
170170
assertFalse(groupMgr.isOnline());
171171
// Verify the group affiliation that's loaded from ZK instead of tables.
172-
assertEquals(newGroup, groupMgr.getRSGroupOfTable(RSGroupInfoManager.RSGROUP_TABLE_NAME));
172+
assertEquals(newGroup, groupMgr.getRSGroupOfTable(RSGroupInfoManagerImpl.RSGROUP_TABLE_NAME));
173173
assertEquals(RSGroupInfo.DEFAULT_GROUP, groupMgr.getRSGroupOfTable(failoverTable));
174174
// Kill final regionserver to see the failover happens for all tables except GROUP table since
175175
// it's group does not have any online RS.
@@ -182,7 +182,7 @@ public boolean evaluate() throws Exception {
182182
return failoverRS.getRegions(failoverTable).size() >= 1;
183183
}
184184
});
185-
Assert.assertEquals(0, failoverRS.getRegions(RSGroupInfoManager.RSGROUP_TABLE_NAME).size());
185+
Assert.assertEquals(0, failoverRS.getRegions(RSGroupInfoManagerImpl.RSGROUP_TABLE_NAME).size());
186186

187187
// Need this for minicluster to shutdown cleanly.
188188
master.stopMaster();

‎hbase-rsgroup/src/test/java/org/apache/hadoop/hbase/rsgroup/VerifyingRSGroupAdminClient.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,7 @@ public VerifyingRSGroupAdminClient(RSGroupAdmin RSGroupAdmin, Configuration conf
5252
throws IOException {
5353
wrapped = RSGroupAdmin;
5454
table = ConnectionFactory.createConnection(conf)
55-
.getTable(RSGroupInfoManager.RSGROUP_TABLE_NAME);
55+
.getTable(RSGroupInfoManagerImpl.RSGROUP_TABLE_NAME);
5656
zkw = new ZKWatcher(conf, this.getClass().getSimpleName(), null);
5757
}
5858

@@ -126,8 +126,8 @@ public void verify() throws IOException {
126126
RSGroupProtos.RSGroupInfo proto =
127127
RSGroupProtos.RSGroupInfo.parseFrom(
128128
result.getValue(
129-
RSGroupInfoManager.META_FAMILY_BYTES,
130-
RSGroupInfoManager.META_QUALIFIER_BYTES));
129+
RSGroupInfoManagerImpl.META_FAMILY_BYTES,
130+
RSGroupInfoManagerImpl.META_QUALIFIER_BYTES));
131131
groupMap.put(proto.getName(), ProtobufUtil.toGroupInfo(proto));
132132
}
133133
Assert.assertEquals(Sets.newHashSet(groupMap.values()),

‎hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManager.java‎ renamed to ‎hbase-server/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManager.java‎

Lines changed: 4 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -15,38 +15,22 @@
1515
* See the License for the specific language governing permissions and
1616
* limitations under the License.
1717
*/
18-
1918
package org.apache.hadoop.hbase.rsgroup;
2019

2120
import java.io.IOException;
2221
import java.util.List;
2322
import java.util.Set;
24-
25-
import org.apache.hadoop.hbase.NamespaceDescriptor;
2623
import org.apache.hadoop.hbase.TableName;
2724
import org.apache.hadoop.hbase.net.Address;
28-
import org.apache.hadoop.hbase.util.Bytes;
2925
import org.apache.yetus.audience.InterfaceAudience;
3026

3127
/**
32-
* Interface used to manage RSGroupInfo storage. An implementation
33-
* has the option to support offline mode.
34-
* See {@link RSGroupBasedLoadBalancer}
28+
* Interface used to manage RSGroupInfo storage. An implementation has the option to support offline
29+
* mode. See {@code RSGroupBasedLoadBalancer}.
3530
*/
3631
@InterfaceAudience.Private
3732
public interface RSGroupInfoManager {
3833

39-
String REASSIGN_WAIT_INTERVAL_KEY = "hbase.rsgroup.reassign.wait";
40-
long DEFAULT_REASSIGN_WAIT_INTERVAL = 30 * 1000L;
41-
42-
//Assigned before user tables
43-
TableName RSGROUP_TABLE_NAME =
44-
TableName.valueOf(NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR, "rsgroup");
45-
String rsGroupZNode = "rsgroup";
46-
byte[] META_FAMILY_BYTES = Bytes.toBytes("m");
47-
byte[] META_QUALIFIER_BYTES = Bytes.toBytes("i");
48-
byte[] ROW_KEY = {0};
49-
5034
void start();
5135

5236
/**
@@ -86,7 +70,6 @@ Set<Address> moveServers(Set<Address> servers, String srcGroup, String dstGroup)
8670

8771
/**
8872
* Set the group membership of a set of tables
89-
*
9073
* @param tableNames set of tables to move
9174
* @param groupName name of group of tables to move to
9275
*/
@@ -104,7 +87,6 @@ Set<Address> moveServers(Set<Address> servers, String srcGroup, String dstGroup)
10487

10588
/**
10689
* Whether the manager is able to fully return group metadata
107-
*
10890
* @return whether the manager is in online mode
10991
*/
11092
boolean isOnline();
@@ -116,8 +98,8 @@ Set<Address> moveServers(Set<Address> servers, String srcGroup, String dstGroup)
11698
* @param srcGroup groupName being moved from
11799
* @param dstGroup groupName being moved to
118100
*/
119-
void moveServersAndTables(Set<Address> servers, Set<TableName> tables,
120-
String srcGroup, String dstGroup) throws IOException;
101+
void moveServersAndTables(Set<Address> servers, Set<TableName> tables, String srcGroup,
102+
String dstGroup) throws IOException;
121103

122104
/**
123105
* Remove decommissioned servers from rsgroup

‎hbase-rsgroup/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManagerImpl.java‎ renamed to ‎hbase-server/src/main/java/org/apache/hadoop/hbase/rsgroup/RSGroupInfoManagerImpl.java‎

Lines changed: 59 additions & 49 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
*/
1818
package org.apache.hadoop.hbase.rsgroup;
1919

20-
import com.google.protobuf.ServiceException;
2120
import java.io.ByteArrayInputStream;
2221
import java.io.IOException;
2322
import java.util.ArrayList;
@@ -34,25 +33,24 @@
3433
import java.util.TreeSet;
3534
import org.apache.hadoop.hbase.Coprocessor;
3635
import org.apache.hadoop.hbase.DoNotRetryIOException;
36+
import org.apache.hadoop.hbase.NamespaceDescriptor;
3737
import org.apache.hadoop.hbase.ServerName;
3838
import org.apache.hadoop.hbase.TableName;
39+
import org.apache.hadoop.hbase.client.AsyncClusterConnection;
40+
import org.apache.hadoop.hbase.client.AsyncTable;
3941
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptorBuilder;
40-
import org.apache.hadoop.hbase.client.Connection;
4142
import org.apache.hadoop.hbase.client.CoprocessorDescriptorBuilder;
4243
import org.apache.hadoop.hbase.client.Delete;
4344
import org.apache.hadoop.hbase.client.Get;
4445
import org.apache.hadoop.hbase.client.Mutation;
4546
import org.apache.hadoop.hbase.client.Put;
4647
import org.apache.hadoop.hbase.client.Result;
4748
import org.apache.hadoop.hbase.client.ResultScanner;
48-
import org.apache.hadoop.hbase.client.Scan;
49-
import org.apache.hadoop.hbase.client.Table;
5049
import org.apache.hadoop.hbase.client.TableDescriptor;
5150
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
5251
import org.apache.hadoop.hbase.constraint.ConstraintException;
5352
import org.apache.hadoop.hbase.coprocessor.MultiRowMutationEndpoint;
5453
import org.apache.hadoop.hbase.exceptions.DeserializationException;
55-
import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
5654
import org.apache.hadoop.hbase.master.MasterServices;
5755
import org.apache.hadoop.hbase.master.ServerListener;
5856
import org.apache.hadoop.hbase.master.TableStateManager;
@@ -62,10 +60,14 @@
6260
import org.apache.hadoop.hbase.procedure2.Procedure;
6361
import org.apache.hadoop.hbase.protobuf.ProtobufMagic;
6462
import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
65-
import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos;
63+
import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto;
64+
import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MultiRowMutationService;
65+
import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MutateRowsRequest;
66+
import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MutateRowsResponse;
6667
import org.apache.hadoop.hbase.protobuf.generated.RSGroupProtos;
6768
import org.apache.hadoop.hbase.regionserver.DisabledRegionSplitPolicy;
6869
import org.apache.hadoop.hbase.util.Bytes;
70+
import org.apache.hadoop.hbase.util.FutureUtils;
6971
import org.apache.hadoop.hbase.util.Threads;
7072
import org.apache.hadoop.hbase.zookeeper.ZKUtil;
7173
import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
@@ -75,6 +77,7 @@
7577
import org.slf4j.Logger;
7678
import org.slf4j.LoggerFactory;
7779

80+
import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
7881
import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
7982
import org.apache.hbase.thirdparty.com.google.common.collect.Maps;
8083
import org.apache.hbase.thirdparty.com.google.common.collect.Sets;
@@ -87,13 +90,13 @@
8790
* RSGroupInfo Map at {@link #rsGroupMap} and a Map of tables to the name of the rsgroup they belong
8891
* too (in {@link #tableMap}). These Maps are persisted to the hbase:rsgroup table (and cached in
8992
* zk) on each modification.
90-
* <p>
93+
* <p/>
9194
* Mutations on state are synchronized but reads can continue without having to wait on an instance
9295
* monitor, mutations do wholesale replace of the Maps on update -- Copy-On-Write; the local Maps of
9396
* state are read-only, just-in-case (see flushConfig).
94-
* <p>
97+
* <p/>
9598
* Reads must not block else there is a danger we'll deadlock.
96-
* <p>
99+
* <p/>
97100
* Clients of this class, the {@link RSGroupAdminEndpoint} for example, want to query and then act
98101
* on the results of the query modifying cache in zookeeper without another thread making
99102
* intermediate modifications. These clients synchronize on the 'this' instance so no other has
@@ -103,6 +106,24 @@
103106
final class RSGroupInfoManagerImpl implements RSGroupInfoManager {
104107
private static final Logger LOG = LoggerFactory.getLogger(RSGroupInfoManagerImpl.class);
105108

109+
private static final String REASSIGN_WAIT_INTERVAL_KEY = "hbase.rsgroup.reassign.wait";
110+
private static final long DEFAULT_REASSIGN_WAIT_INTERVAL = 30 * 1000L;
111+
112+
// Assigned before user tables
113+
@VisibleForTesting
114+
static final TableName RSGROUP_TABLE_NAME =
115+
TableName.valueOf(NamespaceDescriptor.SYSTEM_NAMESPACE_NAME_STR, "rsgroup");
116+
117+
private static final String RS_GROUP_ZNODE = "rsgroup";
118+
119+
@VisibleForTesting
120+
static final byte[] META_FAMILY_BYTES = Bytes.toBytes("m");
121+
122+
@VisibleForTesting
123+
static final byte[] META_QUALIFIER_BYTES = Bytes.toBytes("i");
124+
125+
private static final byte[] ROW_KEY = { 0 };
126+
106127
/** Table descriptor for <code>hbase:rsgroup</code> catalog table */
107128
private static final TableDescriptor RSGROUP_TABLE_DESC;
108129
static {
@@ -125,7 +146,7 @@ final class RSGroupInfoManagerImpl implements RSGroupInfoManager {
125146
private volatile Map<TableName, String> tableMap = Collections.emptyMap();
126147

127148
private final MasterServices masterServices;
128-
private final Connection conn;
149+
private final AsyncClusterConnection conn;
129150
private final ZKWatcher watcher;
130151
private final RSGroupStartupWorker rsGroupStartupWorker;
131152
// contains list of groups that were last flushed to persistent store
@@ -136,7 +157,7 @@ final class RSGroupInfoManagerImpl implements RSGroupInfoManager {
136157
private RSGroupInfoManagerImpl(MasterServices masterServices) throws IOException {
137158
this.masterServices = masterServices;
138159
this.watcher = masterServices.getZooKeeper();
139-
this.conn = masterServices.getConnection();
160+
this.conn = masterServices.getAsyncClusterConnection();
140161
this.rsGroupStartupWorker = new RSGroupStartupWorker();
141162
}
142163

@@ -349,25 +370,25 @@ public synchronized void removeServers(Set<Address> servers) throws IOException
349370
}
350371
}
351372

352-
List<RSGroupInfo> retrieveGroupListFromGroupTable() throws IOException {
373+
private List<RSGroupInfo> retrieveGroupListFromGroupTable() throws IOException {
353374
List<RSGroupInfo> rsGroupInfoList = Lists.newArrayList();
354-
try (Table table = conn.getTable(RSGROUP_TABLE_NAME);
355-
ResultScanner scanner = table.getScanner(new Scan())) {
375+
AsyncTable<?> table = conn.getTable(RSGROUP_TABLE_NAME);
376+
try (ResultScanner scanner = table.getScanner(META_FAMILY_BYTES, META_QUALIFIER_BYTES)) {
356377
for (Result result;;) {
357378
result = scanner.next();
358379
if (result == null) {
359380
break;
360381
}
361382
RSGroupProtos.RSGroupInfo proto = RSGroupProtos.RSGroupInfo
362-
.parseFrom(result.getValue(META_FAMILY_BYTES, META_QUALIFIER_BYTES));
383+
.parseFrom(result.getValue(META_FAMILY_BYTES, META_QUALIFIER_BYTES));
363384
rsGroupInfoList.add(ProtobufUtil.toGroupInfo(proto));
364385
}
365386
}
366387
return rsGroupInfoList;
367388
}
368389

369-
List<RSGroupInfo> retrieveGroupListFromZookeeper() throws IOException {
370-
String groupBasePath = ZNodePaths.joinZNode(watcher.getZNodePaths().baseZNode, rsGroupZNode);
390+
private List<RSGroupInfo> retrieveGroupListFromZookeeper() throws IOException {
391+
String groupBasePath = ZNodePaths.joinZNode(watcher.getZNodePaths().baseZNode, RS_GROUP_ZNODE);
371392
List<RSGroupInfo> RSGroupInfoList = Lists.newArrayList();
372393
// Overwrite any info stored by table, this takes precedence
373394
try {
@@ -519,7 +540,8 @@ private synchronized void flushConfig(Map<String, RSGroupInfo> newGroupMap) thro
519540
resetRSGroupAndTableMaps(newGroupMap, newTableMap);
520541

521542
try {
522-
String groupBasePath = ZNodePaths.joinZNode(watcher.getZNodePaths().baseZNode, rsGroupZNode);
543+
String groupBasePath =
544+
ZNodePaths.joinZNode(watcher.getZNodePaths().baseZNode, RS_GROUP_ZNODE);
523545
ZKUtil.createAndFailSilent(watcher, groupBasePath, ProtobufMagic.PB_MAGIC);
524546

525547
List<ZKUtil.ZKUtilOp> zkOps = new ArrayList<>(newGroupMap.size());
@@ -702,11 +724,8 @@ private boolean waitForGroupTableOnline() {
702724
createRSGroupTable();
703725
}
704726
// try reading from the table
705-
try (Table table = conn.getTable(RSGROUP_TABLE_NAME)) {
706-
table.get(new Get(ROW_KEY));
707-
}
708-
LOG.info(
709-
"RSGroup table=" + RSGROUP_TABLE_NAME + " is online, refreshing cached information");
727+
FutureUtils.get(conn.getTable(RSGROUP_TABLE_NAME).get(new Get(ROW_KEY)));
728+
LOG.info("RSGroup table={} is online, refreshing cached information", RSGROUP_TABLE_NAME);
710729
RSGroupInfoManagerImpl.this.refresh(true);
711730
online = true;
712731
// flush any inconsistencies between ZK and HTable
@@ -748,8 +767,8 @@ private void createRSGroupTable() throws IOException {
748767
} else {
749768
Procedure<?> result = masterServices.getMasterProcedureExecutor().getResult(procId);
750769
if (result != null && result.isFailed()) {
751-
throw new IOException(
752-
"Failed to create group table. " + MasterProcedureUtil.unwrapRemoteIOException(result));
770+
throw new IOException("Failed to create group table. " +
771+
MasterProcedureUtil.unwrapRemoteIOException(result));
753772
}
754773
}
755774
}
@@ -764,33 +783,24 @@ private static boolean isMasterRunning(MasterServices masterServices) {
764783
}
765784

766785
private void multiMutate(List<Mutation> mutations) throws IOException {
767-
try (Table table = conn.getTable(RSGROUP_TABLE_NAME)) {
768-
CoprocessorRpcChannel channel = table.coprocessorService(ROW_KEY);
769-
MultiRowMutationProtos.MutateRowsRequest.Builder mmrBuilder =
770-
MultiRowMutationProtos.MutateRowsRequest.newBuilder();
771-
for (Mutation mutation : mutations) {
772-
if (mutation instanceof Put) {
773-
mmrBuilder.addMutationRequest(org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(
774-
org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.MutationType.PUT,
775-
mutation));
776-
} else if (mutation instanceof Delete) {
777-
mmrBuilder.addMutationRequest(org.apache.hadoop.hbase.protobuf.ProtobufUtil.toMutation(
778-
org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutationProto.MutationType.DELETE,
779-
mutation));
780-
} else {
781-
throw new DoNotRetryIOException(
786+
MutateRowsRequest.Builder builder = MutateRowsRequest.newBuilder();
787+
for (Mutation mutation : mutations) {
788+
if (mutation instanceof Put) {
789+
builder
790+
.addMutationRequest(ProtobufUtil.toMutation(MutationProto.MutationType.PUT, mutation));
791+
} else if (mutation instanceof Delete) {
792+
builder.addMutationRequest(
793+
ProtobufUtil.toMutation(MutationProto.MutationType.DELETE, mutation));
794+
} else {
795+
throw new DoNotRetryIOException(
782796
"multiMutate doesn't support " + mutation.getClass().getName());
783-
}
784-
}
785-
786-
MultiRowMutationProtos.MultiRowMutationService.BlockingInterface service =
787-
MultiRowMutationProtos.MultiRowMutationService.newBlockingStub(channel);
788-
try {
789-
service.mutateRows(null, mmrBuilder.build());
790-
} catch (ServiceException ex) {
791-
ProtobufUtil.toIOException(ex);
792797
}
793798
}
799+
MutateRowsRequest request = builder.build();
800+
AsyncTable<?> table = conn.getTable(RSGROUP_TABLE_NAME);
801+
FutureUtils.get(table.<MultiRowMutationService, MutateRowsResponse> coprocessorService(
802+
MultiRowMutationService::newStub,
803+
(stub, controller, done) -> stub.mutateRows(controller, request, done), ROW_KEY));
794804
}
795805

796806
private void checkGroupName(String groupName) throws ConstraintException {

0 commit comments

Comments
 (0)