Apache9 commented on a change in pull request #3911:
URL: https://github.com/apache/hbase/pull/3911#discussion_r762403092
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java
##########
@@ -242,21 +276,150 @@ public boolean isAborted() {
if (children == null) {
return Collections.emptyList();
}
+
+ Configuration conf = HBaseConfiguration.create();
+
+ /** if use other balancer, return all regionservers */
+ if (!conf.get(HConstants.HBASE_MASTER_LOADBALANCER_CLASS)
+ .equals(RSGroupBasedLoadBalancer.class.getName())
+ || hostServerName == null) {
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Use replication random choose policy...");
+ }
+ return parseServerNameFromList(children);
+ } else {
+ /** if use rsgroup balancer,
+ * just return regionservers belong to the same rsgroup or default
rsgroup */
+ if(LOG.isDebugEnabled()) {
+ LOG.debug("Use replication rsgroup choose policy...");
+ }
+ Map<String, String> serverNameHostPortMapping = new HashMap<>();
+ for (String serverName : children) {
+ String mappingKey =
+ serverName.split(",")[0] + ServerName.SERVERNAME_SEPARATOR +
serverName.split(",")[1];
+ serverNameHostPortMapping.put(mappingKey, serverName);
+ }
+
+ String groupName = null;
+ RSGroupInfo rsGroupInfo = null;
+ try {
+ rsGroupInfo = getRSGroupInfoOfServer(conn.toConnection(),
hostServerName.getAddress());
+ }catch (IOException e) {
+ e.printStackTrace();
+ LOG.error("rsGroupInfo error!", e);
+ }
+ if (rsGroupInfo != null) {
+ groupName = rsGroupInfo.getName();
+ }
+ try {
+ List<ServerName> serverList =
+ getGroupServerListFromTargetZkCluster(groupName, zkw,
serverNameHostPortMapping);
+ if (serverList.size() > 0) {
+ // if target cluster open group balancer, serverList must has
server(s)
+ LOG.debug("group list > 0");
+ threadLocal.get().getAndSet(true);
+ return serverList;
+ }
+ else {
+ // if not, choose sinkers from all regionservers
+ LOG.debug("target group list <= 0");
+ return parseServerNameFromList(children);
+ }
+ }catch (IOException | KeeperException e) {
+ LOG.error("Get server list from target zk error", e);
+ return Collections.emptyList();
+ }
+ }
+ }
+
+ protected List<ServerName> parseServerNameFromList(List<String> children) {
+ if (children == null) {
+ return Collections.emptyList();
+ }
+ StringBuffer sb = new StringBuffer();
List<ServerName> addresses = new ArrayList<>(children.size());
for (String child : children) {
addresses.add(ServerName.parseServerName(child));
+ sb.append(ServerName.parseServerName(child)).append("/");
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug(
+ "Find " + addresses.size() + " child znodes from target cluster zk. "
+ sb.toString());
}
return addresses;
}
+ protected List<ServerName> getGroupServerListFromTargetZkCluster(String
groupName,
Review comment:
On master branch we do not use zk for storing the rs group any more...
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java
##########
@@ -77,9 +90,13 @@
*/
public static final float DEFAULT_REPLICATION_SOURCE_RATIO = 0.5f;
+ static final float DEFAULT_REPLICATION_SOURCE_GROUP_RATIO = 1f;
+
// Ratio of total number of potential peer region servers to be used
private float ratio;
+ private float groupRatio;
Review comment:
Yes, please add some explaination.
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java
##########
@@ -88,6 +105,17 @@
private List<ServerName> sinkServers = new ArrayList<>(0);
+ private static ThreadLocal<AtomicBoolean> threadLocal = new
ThreadLocal<AtomicBoolean>() {
Review comment:
We need to use AtomicBoolean for a thread local variable?
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/ReplicationService.java
##########
@@ -32,6 +32,7 @@
*/
@InterfaceAudience.Private
public interface ReplicationService {
+
Review comment:
Please avoid touching unnecessary file.
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java
##########
@@ -242,21 +276,150 @@ public boolean isAborted() {
if (children == null) {
return Collections.emptyList();
}
+
+ Configuration conf = HBaseConfiguration.create();
+
+ /** if use other balancer, return all regionservers */
+ if (!conf.get(HConstants.HBASE_MASTER_LOADBALANCER_CLASS)
Review comment:
And another problem is do we need to check the configuration here? I
think it is the version at the source cluster, but we need to check the target
cluster?
##########
File path:
hbase-server/src/main/java/org/apache/hadoop/hbase/replication/HBaseReplicationEndpoint.java
##########
@@ -242,21 +276,150 @@ public boolean isAborted() {
if (children == null) {
return Collections.emptyList();
}
+
+ Configuration conf = HBaseConfiguration.create();
+
+ /** if use other balancer, return all regionservers */
+ if (!conf.get(HConstants.HBASE_MASTER_LOADBALANCER_CLASS)
Review comment:
On master this is not the case now. The balancer will always be a
RSGroupBasedLoadBalancer, if we do not enable rs group feature,
DisabledRSGroupInfoManager will be used and there will be only one group, which
will act as there is no rs group. So I think here we should check for
RSGroupUtil.isRSGroupEnabled.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
