Skip to content

Commit 5d872d3

Browse files
Apache9Apache9
committed
HBASE-21526 Use AsyncClusterConnection in ServerManager for getRsAdmin
1 parent 2446f00 commit 5d872d3

File tree

7 files changed

+244
-87
lines changed

7 files changed

+244
-87
lines changed

‎hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncClusterConnection.java

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

20+
import org.apache.hadoop.hbase.ServerName;
2021
import org.apache.hadoop.hbase.ipc.RpcClient;
2122
import org.apache.yetus.audience.InterfaceAudience;
2223

@@ -26,6 +27,11 @@
2627
@InterfaceAudience.Private
2728
public interface AsyncClusterConnection extends AsyncConnection {
2829

30+
/**
31+
* Get the admin service for the given region server.
32+
*/
33+
AsyncRegionServerAdmin getRegionServerAdmin(ServerName serverName);
34+
2935
/**
3036
* Get the nonce generator for this connection.
3137
*/

‎hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncConnectionImpl.java

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -379,4 +379,9 @@ public void clearRegionLocationCache() {
379379
Optional<MetricsConnection> getConnectionMetrics() {
380380
return metrics;
381381
}
382+
383+
@Override
384+
public AsyncRegionServerAdmin getRegionServerAdmin(ServerName serverName) {
385+
return new AsyncRegionServerAdmin(serverName, this);
386+
}
382387
}

‎hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncRegionServerAdmin.java

Lines changed: 210 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,210 @@
1+
/**
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing, software
13+
* distributed under the License is distributed on an "AS IS" BASIS,
14+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
* See the License for the specific language governing permissions and
16+
* limitations under the License.
17+
*/
18+
package org.apache.hadoop.hbase.client;
19+
20+
import java.io.IOException;
21+
import java.util.concurrent.CompletableFuture;
22+
import org.apache.hadoop.hbase.ServerName;
23+
import org.apache.hadoop.hbase.ipc.HBaseRpcController;
24+
import org.apache.yetus.audience.InterfaceAudience;
25+
26+
import org.apache.hbase.thirdparty.com.google.protobuf.RpcCallback;
27+
import org.apache.hbase.thirdparty.com.google.protobuf.RpcController;
28+
29+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
30+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesRequest;
31+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearCompactionQueuesResponse;
32+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearRegionBlockCacheRequest;
33+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearRegionBlockCacheResponse;
34+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CloseRegionRequest;
35+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CloseRegionResponse;
36+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CompactRegionRequest;
37+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CompactRegionResponse;
38+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CompactionSwitchRequest;
39+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CompactionSwitchResponse;
40+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ExecuteProceduresRequest;
41+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ExecuteProceduresResponse;
42+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.FlushRegionRequest;
43+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.FlushRegionResponse;
44+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetOnlineRegionRequest;
45+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetOnlineRegionResponse;
46+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetRegionInfoRequest;
47+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetRegionInfoResponse;
48+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetRegionLoadRequest;
49+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetRegionLoadResponse;
50+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetServerInfoRequest;
51+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetServerInfoResponse;
52+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetStoreFileRequest;
53+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetStoreFileResponse;
54+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.OpenRegionRequest;
55+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.OpenRegionResponse;
56+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ReplicateWALEntryRequest;
57+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ReplicateWALEntryResponse;
58+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.RollWALWriterRequest;
59+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.RollWALWriterResponse;
60+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.StopServerRequest;
61+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.StopServerResponse;
62+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationRequest;
63+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationResponse;
64+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateFavoredNodesRequest;
65+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateFavoredNodesResponse;
66+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.WarmupRegionRequest;
67+
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.WarmupRegionResponse;
68+
import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.GetSpaceQuotaSnapshotsRequest;
69+
import org.apache.hadoop.hbase.shaded.protobuf.generated.QuotaProtos.GetSpaceQuotaSnapshotsResponse;
70+
71+
/**
72+
* A simple wrapper of the {@link AdminService} for a region server, which returns a
73+
* {@link CompletableFuture}. This is easier to use, as if you use the raw protobuf interface, you
74+
* need to get the result from the {@link RpcCallback}, and if there is an exception, you need to
75+
* get it from the {@link RpcController} passed in.
76+
* <p/>
77+
* Notice that there is no retry, and this is intentional. We have different retry for different
78+
* usage for now, if later we want to unify them, we can move the retry logic into this class.
79+
*/
80+
@InterfaceAudience.Private
81+
public class AsyncRegionServerAdmin {
82+
83+
private final ServerName server;
84+
85+
private final AsyncConnectionImpl conn;
86+
87+
AsyncRegionServerAdmin(ServerName server, AsyncConnectionImpl conn) {
88+
this.server = server;
89+
this.conn = conn;
90+
}
91+
92+
@FunctionalInterface
93+
private interface RpcCall<RESP> {
94+
void call(AdminService.Interface stub, HBaseRpcController controller, RpcCallback<RESP> done);
95+
}
96+
97+
private <RESP> CompletableFuture<RESP> call(RpcCall<RESP> rpcCall) {
98+
CompletableFuture<RESP> future = new CompletableFuture<>();
99+
HBaseRpcController controller = conn.rpcControllerFactory.newController();
100+
try {
101+
rpcCall.call(conn.getAdminStub(server), controller, new RpcCallback<RESP>() {
102+
103+
@Override
104+
public void run(RESP resp) {
105+
if (controller.failed()) {
106+
future.completeExceptionally(controller.getFailed());
107+
} else {
108+
future.complete(resp);
109+
}
110+
}
111+
});
112+
} catch (IOException e) {
113+
future.completeExceptionally(e);
114+
}
115+
return future;
116+
}
117+
118+
public CompletableFuture<GetRegionInfoResponse> getRegionInfo(GetRegionInfoRequest request) {
119+
return call((stub, controller, done) -> stub.getRegionInfo(controller, request, done));
120+
}
121+
122+
public CompletableFuture<GetStoreFileResponse> getStoreFile(GetStoreFileRequest request) {
123+
return call((stub, controller, done) -> stub.getStoreFile(controller, request, done));
124+
}
125+
126+
public CompletableFuture<GetOnlineRegionResponse> getOnlineRegion(
127+
GetOnlineRegionRequest request) {
128+
return call((stub, controller, done) -> stub.getOnlineRegion(controller, request, done));
129+
}
130+
131+
public CompletableFuture<OpenRegionResponse> openRegion(OpenRegionRequest request) {
132+
return call((stub, controller, done) -> stub.openRegion(controller, request, done));
133+
}
134+
135+
public CompletableFuture<WarmupRegionResponse> warmupRegion(WarmupRegionRequest request) {
136+
return call((stub, controller, done) -> stub.warmupRegion(controller, request, done));
137+
}
138+
139+
public CompletableFuture<CloseRegionResponse> closeRegion(CloseRegionRequest request) {
140+
return call((stub, controller, done) -> stub.closeRegion(controller, request, done));
141+
}
142+
143+
public CompletableFuture<FlushRegionResponse> flushRegion(FlushRegionRequest request) {
144+
return call((stub, controller, done) -> stub.flushRegion(controller, request, done));
145+
}
146+
147+
public CompletableFuture<CompactionSwitchResponse> compactionSwitch(
148+
CompactionSwitchRequest request) {
149+
return call((stub, controller, done) -> stub.compactionSwitch(controller, request, done));
150+
}
151+
152+
public CompletableFuture<CompactRegionResponse> compactRegion(CompactRegionRequest request) {
153+
return call((stub, controller, done) -> stub.compactRegion(controller, request, done));
154+
}
155+
156+
public CompletableFuture<ReplicateWALEntryResponse> replicateWALEntry(
157+
ReplicateWALEntryRequest request) {
158+
return call((stub, controller, done) -> stub.replicateWALEntry(controller, request, done));
159+
}
160+
161+
public CompletableFuture<ReplicateWALEntryResponse> replay(ReplicateWALEntryRequest request) {
162+
return call((stub, controller, done) -> stub.replay(controller, request, done));
163+
}
164+
165+
public CompletableFuture<RollWALWriterResponse> rollWALWriter(RollWALWriterRequest request) {
166+
return call((stub, controller, done) -> stub.rollWALWriter(controller, request, done));
167+
}
168+
169+
public CompletableFuture<GetServerInfoResponse> getServerInfo(GetServerInfoRequest request) {
170+
return call((stub, controller, done) -> stub.getServerInfo(controller, request, done));
171+
}
172+
173+
public CompletableFuture<StopServerResponse> stopServer(StopServerRequest request) {
174+
return call((stub, controller, done) -> stub.stopServer(controller, request, done));
175+
}
176+
177+
public CompletableFuture<UpdateFavoredNodesResponse> updateFavoredNodes(
178+
UpdateFavoredNodesRequest request) {
179+
return call((stub, controller, done) -> stub.updateFavoredNodes(controller, request, done));
180+
}
181+
182+
public CompletableFuture<UpdateConfigurationResponse> updateConfiguration(
183+
UpdateConfigurationRequest request) {
184+
return call((stub, controller, done) -> stub.updateConfiguration(controller, request, done));
185+
}
186+
187+
public CompletableFuture<GetRegionLoadResponse> getRegionLoad(GetRegionLoadRequest request) {
188+
return call((stub, controller, done) -> stub.getRegionLoad(controller, request, done));
189+
}
190+
191+
public CompletableFuture<ClearCompactionQueuesResponse> clearCompactionQueues(
192+
ClearCompactionQueuesRequest request) {
193+
return call((stub, controller, done) -> stub.clearCompactionQueues(controller, request, done));
194+
}
195+
196+
public CompletableFuture<ClearRegionBlockCacheResponse> clearRegionBlockCache(
197+
ClearRegionBlockCacheRequest request) {
198+
return call((stub, controller, done) -> stub.clearRegionBlockCache(controller, request, done));
199+
}
200+
201+
public CompletableFuture<GetSpaceQuotaSnapshotsResponse> getSpaceQuotaSnapshots(
202+
GetSpaceQuotaSnapshotsRequest request) {
203+
return call((stub, controller, done) -> stub.getSpaceQuotaSnapshots(controller, request, done));
204+
}
205+
206+
public CompletableFuture<ExecuteProceduresResponse> executeProcedures(
207+
ExecuteProceduresRequest request) {
208+
return call((stub, controller, done) -> stub.executeProcedures(controller, request, done));
209+
}
210+
}

‎hbase-common/src/main/java/org/apache/hadoop/hbase/util/FutureUtils.java

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -193,4 +193,4 @@ public static <T> CompletableFuture<T> failedFuture(Throwable e) {
193193
future.completeExceptionally(e);
194194
return future;
195195
}
196-
}
196+
}

‎hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java

Lines changed: 17 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -202,6 +202,7 @@
202202
import org.apache.hadoop.hbase.util.Bytes;
203203
import org.apache.hadoop.hbase.util.CompressionTest;
204204
import org.apache.hadoop.hbase.util.EncryptionTest;
205+
import org.apache.hadoop.hbase.util.FutureUtils;
205206
import org.apache.hadoop.hbase.util.HBaseFsck;
206207
import org.apache.hadoop.hbase.util.HFileArchiveUtil;
207208
import org.apache.hadoop.hbase.util.HasThread;
@@ -234,6 +235,7 @@
234235
import org.apache.hbase.thirdparty.com.google.common.collect.Maps;
235236

236237
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
238+
import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
237239
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetRegionInfoResponse.CompactionState;
238240
import org.apache.hadoop.hbase.shaded.protobuf.generated.SnapshotProtos.SnapshotDescription;
239241
import org.apache.hadoop.hbase.shaded.protobuf.generated.WALProtos;
@@ -1961,6 +1963,15 @@ protected String getDescription() {
19611963
});
19621964
}
19631965

1966+
private void warmUpRegion(ServerName server, RegionInfo region) {
1967+
FutureUtils.addListener(asyncClusterConnection.getRegionServerAdmin(server)
1968+
.warmupRegion(RequestConverter.buildWarmupRegionRequest(region)), (r, e) -> {
1969+
if (e != null) {
1970+
LOG.warn("Failed to warm up region {} on server {}", region, server, e);
1971+
}
1972+
});
1973+
}
1974+
19641975
// Public so can be accessed by tests. Blocks until move is done.
19651976
// Replace with an async implementation from which you can get
19661977
// a success/failure result.
@@ -2031,11 +2042,12 @@ public void move(final byte[] encodedRegionName, byte[] destServerName) throws H
20312042
}
20322043

20332044
TransitRegionStateProcedure proc =
2034-
this.assignmentManager.createMoveRegionProcedure(rp.getRegionInfo(), rp.getDestination());
2035-
// Warmup the region on the destination before initiating the move. this call
2036-
// is synchronous and takes some time. doing it before the source region gets
2037-
// closed
2038-
serverManager.sendRegionWarmup(rp.getDestination(), hri);
2045+
this.assignmentManager.createMoveRegionProcedure(rp.getRegionInfo(), rp.getDestination());
2046+
// Warmup the region on the destination before initiating the move.
2047+
// A region server could reject the close request because it either does not
2048+
// have the specified region or the region is being split.
2049+
warmUpRegion(rp.getDestination(), hri);
2050+
20392051
LOG.info(getClientIdAuditPrefix() + " move " + rp + ", running balancer");
20402052
Future<byte[]> future = ProcedureSyncWait.submitProcedure(this.procedureExecutor, proc);
20412053
try {

0 commit comments

Comments
 (0)