You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hbase.apache.org by sy...@apache.org on 2017/04/26 22:52:24 UTC
[17/40] hbase git commit: HBASE-17915 Implement async replication
admin methods
HBASE-17915 Implement async replication admin methods
Project: http://git-wip-us.apache.org/repos/asf/hbase/repo
Commit: http://git-wip-us.apache.org/repos/asf/hbase/commit/40cc666a
Tree: http://git-wip-us.apache.org/repos/asf/hbase/tree/40cc666a
Diff: http://git-wip-us.apache.org/repos/asf/hbase/diff/40cc666a
Branch: refs/heads/hbase-12439
Commit: 40cc666ac984e846a8c7105b771ce6bec90c4ad3
Parents: b1ef8dd
Author: Guanghao Zhang <zg...@apache.org>
Authored: Thu Apr 20 18:13:03 2017 +0800
Committer: Guanghao Zhang <zg...@apache.org>
Committed: Thu Apr 20 18:13:03 2017 +0800
----------------------------------------------------------------------
.../apache/hadoop/hbase/client/AsyncAdmin.java | 92 ++++
.../hadoop/hbase/client/AsyncHBaseAdmin.java | 249 +++++++++--
.../apache/hadoop/hbase/client/HBaseAdmin.java | 58 +--
.../replication/ReplicationSerDeHelper.java | 67 +++
.../hadoop/hbase/client/TestAsyncAdminBase.java | 2 +-
.../client/TestAsyncReplicationAdminApi.java | 416 +++++++++++++++++++
6 files changed, 802 insertions(+), 82 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hbase/blob/40cc666a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java
index 270f28f..5d2955f 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java
@@ -18,6 +18,8 @@
package org.apache.hadoop.hbase.client;
import java.util.List;
+import java.util.Collection;
+import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.regex.Pattern;
@@ -30,6 +32,9 @@ import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.quotas.QuotaFilter;
import org.apache.hadoop.hbase.quotas.QuotaSettings;
+import org.apache.hadoop.hbase.client.replication.TableCFs;
+import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
+import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
import org.apache.hadoop.hbase.util.Pair;
/**
@@ -481,4 +486,91 @@ public interface AsyncAdmin {
* @return the QuotaSetting list, which wrapped by a CompletableFuture.
*/
CompletableFuture<List<QuotaSettings>> getQuota(QuotaFilter filter);
+
+ /**
+ * Add a new replication peer for replicating data to slave cluster
+ * @param peerId a short name that identifies the peer
+ * @param peerConfig configuration for the replication slave cluster
+ */
+ CompletableFuture<Void> addReplicationPeer(final String peerId,
+ final ReplicationPeerConfig peerConfig);
+
+ /**
+ * Remove a peer and stop the replication
+ * @param peerId a short name that identifies the peer
+ */
+ CompletableFuture<Void> removeReplicationPeer(final String peerId);
+
+ /**
+ * Restart the replication stream to the specified peer
+ * @param peerId a short name that identifies the peer
+ */
+ CompletableFuture<Void> enableReplicationPeer(final String peerId);
+
+ /**
+ * Stop the replication stream to the specified peer
+ * @param peerId a short name that identifies the peer
+ */
+ CompletableFuture<Void> disableReplicationPeer(final String peerId);
+
+ /**
+ * Returns the configured ReplicationPeerConfig for the specified peer
+ * @param peerId a short name that identifies the peer
+ * @return ReplicationPeerConfig for the peer wrapped by a {@link CompletableFuture}.
+ */
+ CompletableFuture<ReplicationPeerConfig> getReplicationPeerConfig(final String peerId);
+
+ /**
+ * Update the peerConfig for the specified peer
+ * @param peerId a short name that identifies the peer
+ * @param peerConfig new config for the peer
+ */
+ CompletableFuture<Void> updateReplicationPeerConfig(final String peerId,
+ final ReplicationPeerConfig peerConfig);
+
+ /**
+ * Append the replicable table-cf config of the specified peer
+ * @param id a short that identifies the cluster
+ * @param tableCfs A map from tableName to column family names
+ */
+ CompletableFuture<Void> appendReplicationPeerTableCFs(String id,
+ Map<TableName, ? extends Collection<String>> tableCfs);
+
+ /**
+ * Remove some table-cfs from config of the specified peer
+ * @param id a short name that identifies the cluster
+ * @param tableCfs A map from tableName to column family names
+ */
+ CompletableFuture<Void> removeReplicationPeerTableCFs(String id,
+ Map<TableName, ? extends Collection<String>> tableCfs);
+
+ /**
+ * Return a list of replication peers.
+ * @return a list of replication peers description. The return value will be wrapped by a
+ * {@link CompletableFuture}.
+ */
+ CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers();
+
+ /**
+ * Return a list of replication peers.
+ * @param regex The regular expression to match peer id
+ * @return a list of replication peers description. The return value will be wrapped by a
+ * {@link CompletableFuture}.
+ */
+ CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(String regex);
+
+ /**
+ * Return a list of replication peers.
+ * @param pattern The compiled regular expression to match peer id
+ * @return a list of replication peers description. The return value will be wrapped by a
+ * {@link CompletableFuture}.
+ */
+ CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(Pattern pattern);
+
+ /**
+ * Find all table and column families that are replicated from this cluster
+ * @return the replicated table-cfs list of this cluster. The return value will be wrapped by a
+ * {@link CompletableFuture}.
+ */
+ CompletableFuture<List<TableCFs>> listReplicatedTableCFs();
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/40cc666a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java
index 180cd19..eae4089 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java
@@ -22,16 +22,21 @@ import static org.apache.hadoop.hbase.TableName.META_TABLE_NAME;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiConsumer;
import java.util.regex.Pattern;
+import java.util.stream.Collectors;
import com.google.common.annotations.VisibleForTesting;
+
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.HColumnDescriptor;
@@ -54,11 +59,16 @@ import org.apache.hadoop.hbase.classification.InterfaceStability;
import org.apache.hadoop.hbase.client.AsyncRpcRetryingCallerFactory.AdminRequestCallerBuilder;
import org.apache.hadoop.hbase.client.AsyncRpcRetryingCallerFactory.MasterRequestCallerBuilder;
import org.apache.hadoop.hbase.client.Scan.ReadType;
+import org.apache.hadoop.hbase.client.replication.ReplicationSerDeHelper;
+import org.apache.hadoop.hbase.client.replication.TableCFs;
import org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.ipc.HBaseRpcController;
import org.apache.hadoop.hbase.quotas.QuotaFilter;
import org.apache.hadoop.hbase.quotas.QuotaSettings;
import org.apache.hadoop.hbase.quotas.QuotaTableUtil;
+import org.apache.hadoop.hbase.replication.ReplicationException;
+import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
+import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
import org.apache.hadoop.hbase.shaded.com.google.protobuf.RpcCallback;
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
@@ -121,6 +131,20 @@ import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.TruncateTa
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.TruncateTableResponse;
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.UnassignRegionRequest;
import org.apache.hadoop.hbase.shaded.protobuf.generated.MasterProtos.UnassignRegionResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.AddReplicationPeerRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.AddReplicationPeerResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.DisableReplicationPeerRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.DisableReplicationPeerResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.EnableReplicationPeerRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.EnableReplicationPeerResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.GetReplicationPeerConfigRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.GetReplicationPeerConfigResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.ListReplicationPeersRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.ListReplicationPeersResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.RemoveReplicationPeerRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.RemoveReplicationPeerResponse;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.UpdateReplicationPeerConfigRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos.UpdateReplicationPeerConfigResponse;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.ForeignExceptionUtil;
import org.apache.hadoop.hbase.util.Pair;
@@ -1155,42 +1179,209 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
}
@Override
- public CompletableFuture<Void> setQuota(QuotaSettings quota){
- return this.<Void> newMasterCaller()
- .action((controller, stub) -> this.<SetQuotaRequest, SetQuotaResponse, Void> call(
- controller, stub, QuotaSettings.buildSetQuotaRequestProto(quota),
- (s, c, req, done) -> s.setQuota(c, req, done), (resp) -> null))
- .call();
+ public CompletableFuture<Void> setQuota(QuotaSettings quota) {
+ return this
+ .<Void> newMasterCaller()
+ .action(
+ (controller, stub) -> this.<SetQuotaRequest, SetQuotaResponse, Void> call(controller,
+ stub, QuotaSettings.buildSetQuotaRequestProto(quota),
+ (s, c, req, done) -> s.setQuota(c, req, done), (resp) -> null)).call();
}
@Override
public CompletableFuture<List<QuotaSettings>> getQuota(QuotaFilter filter) {
CompletableFuture<List<QuotaSettings>> future = new CompletableFuture<>();
Scan scan = QuotaTableUtil.makeScan(filter);
- this.connection.getRawTableBuilder(QuotaTableUtil.QUOTA_TABLE_NAME).build().scan(scan,
- new RawScanResultConsumer() {
- List<QuotaSettings> settings = new ArrayList<>();
-
- @Override
- public void onNext(Result[] results, ScanController controller) {
- for (Result result : results) {
- try {
- QuotaTableUtil.parseResultToCollection(result, settings);
- } catch (IOException e) {
- controller.terminate();
- future.completeExceptionally(e);
+ this.connection.getRawTableBuilder(QuotaTableUtil.QUOTA_TABLE_NAME).build()
+ .scan(scan, new RawScanResultConsumer() {
+ List<QuotaSettings> settings = new ArrayList<>();
+
+ @Override
+ public void onNext(Result[] results, ScanController controller) {
+ for (Result result : results) {
+ try {
+ QuotaTableUtil.parseResultToCollection(result, settings);
+ } catch (IOException e) {
+ controller.terminate();
+ future.completeExceptionally(e);
+ }
}
}
- }
- @Override
- public void onError(Throwable error) {
- future.completeExceptionally(error);
+ @Override
+ public void onError(Throwable error) {
+ future.completeExceptionally(error);
+ }
+
+ @Override
+ public void onComplete() {
+ future.complete(settings);
+ }
+ });
+ return future;
+ }
+
+ public CompletableFuture<Void> addReplicationPeer(String peerId, ReplicationPeerConfig peerConfig) {
+ return this
+ .<Void> newMasterCaller()
+ .action(
+ (controller, stub) -> this
+ .<AddReplicationPeerRequest, AddReplicationPeerResponse, Void> call(controller, stub,
+ RequestConverter.buildAddReplicationPeerRequest(peerId, peerConfig), (s, c, req,
+ done) -> s.addReplicationPeer(c, req, done), (resp) -> null)).call();
+ }
+
+ @Override
+ public CompletableFuture<Void> removeReplicationPeer(String peerId) {
+ return this
+ .<Void> newMasterCaller()
+ .action(
+ (controller, stub) -> this
+ .<RemoveReplicationPeerRequest, RemoveReplicationPeerResponse, Void> call(controller,
+ stub, RequestConverter.buildRemoveReplicationPeerRequest(peerId),
+ (s, c, req, done) -> s.removeReplicationPeer(c, req, done), (resp) -> null)).call();
+ }
+
+ @Override
+ public CompletableFuture<Void> enableReplicationPeer(String peerId) {
+ return this
+ .<Void> newMasterCaller()
+ .action(
+ (controller, stub) -> this
+ .<EnableReplicationPeerRequest, EnableReplicationPeerResponse, Void> call(controller,
+ stub, RequestConverter.buildEnableReplicationPeerRequest(peerId),
+ (s, c, req, done) -> s.enableReplicationPeer(c, req, done), (resp) -> null)).call();
+ }
+
+ @Override
+ public CompletableFuture<Void> disableReplicationPeer(String peerId) {
+ return this
+ .<Void> newMasterCaller()
+ .action(
+ (controller, stub) -> this
+ .<DisableReplicationPeerRequest, DisableReplicationPeerResponse, Void> call(
+ controller, stub, RequestConverter.buildDisableReplicationPeerRequest(peerId), (s,
+ c, req, done) -> s.disableReplicationPeer(c, req, done), (resp) -> null))
+ .call();
+ }
+
+ public CompletableFuture<ReplicationPeerConfig> getReplicationPeerConfig(String peerId) {
+ return this
+ .<ReplicationPeerConfig> newMasterCaller()
+ .action(
+ (controller, stub) -> this
+ .<GetReplicationPeerConfigRequest, GetReplicationPeerConfigResponse, ReplicationPeerConfig> call(
+ controller, stub, RequestConverter.buildGetReplicationPeerConfigRequest(peerId), (
+ s, c, req, done) -> s.getReplicationPeerConfig(c, req, done),
+ (resp) -> ReplicationSerDeHelper.convert(resp.getPeerConfig()))).call();
+ }
+
+ @Override
+ public CompletableFuture<Void> updateReplicationPeerConfig(String peerId,
+ ReplicationPeerConfig peerConfig) {
+ return this
+ .<Void> newMasterCaller()
+ .action(
+ (controller, stub) -> this
+ .<UpdateReplicationPeerConfigRequest, UpdateReplicationPeerConfigResponse, Void> call(
+ controller, stub, RequestConverter.buildUpdateReplicationPeerConfigRequest(peerId,
+ peerConfig), (s, c, req, done) -> s.updateReplicationPeerConfig(c, req, done), (
+ resp) -> null)).call();
+ }
+
+ @Override
+ public CompletableFuture<Void> appendReplicationPeerTableCFs(String id,
+ Map<TableName, ? extends Collection<String>> tableCfs) {
+ if (tableCfs == null) {
+ return failedFuture(new ReplicationException("tableCfs is null"));
+ }
+
+ CompletableFuture<Void> future = new CompletableFuture<Void>();
+ getReplicationPeerConfig(id).whenComplete((peerConfig, error) -> {
+ if (!completeExceptionally(future, error)) {
+ ReplicationSerDeHelper.appendTableCFsToReplicationPeerConfig(tableCfs, peerConfig);
+ updateReplicationPeerConfig(id, peerConfig).whenComplete((result, err) -> {
+ if (!completeExceptionally(future, error)) {
+ future.complete(result);
+ }
+ });
+ }
+ });
+ return future;
+ }
+
+ @Override
+ public CompletableFuture<Void> removeReplicationPeerTableCFs(String id,
+ Map<TableName, ? extends Collection<String>> tableCfs) {
+ if (tableCfs == null) {
+ return failedFuture(new ReplicationException("tableCfs is null"));
+ }
+
+ CompletableFuture<Void> future = new CompletableFuture<Void>();
+ getReplicationPeerConfig(id).whenComplete((peerConfig, error) -> {
+ if (!completeExceptionally(future, error)) {
+ try {
+ ReplicationSerDeHelper.removeTableCFsFromReplicationPeerConfig(tableCfs, peerConfig, id);
+ } catch (ReplicationException e) {
+ future.completeExceptionally(e);
+ return;
}
+ updateReplicationPeerConfig(id, peerConfig).whenComplete((result, err) -> {
+ if (!completeExceptionally(future, error)) {
+ future.complete(result);
+ }
+ });
+ }
+ });
+ return future;
+ }
+
+ @Override
+ public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers() {
+ return listReplicationPeers((Pattern) null);
+ }
- @Override
- public void onComplete() {
- future.complete(settings);
+ @Override
+ public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(String regex) {
+ return listReplicationPeers(Pattern.compile(regex));
+ }
+
+ @Override
+ public CompletableFuture<List<ReplicationPeerDescription>> listReplicationPeers(Pattern pattern) {
+ return this
+ .<List<ReplicationPeerDescription>> newMasterCaller()
+ .action(
+ (controller, stub) -> this
+ .<ListReplicationPeersRequest, ListReplicationPeersResponse, List<ReplicationPeerDescription>> call(
+ controller,
+ stub,
+ RequestConverter.buildListReplicationPeersRequest(pattern),
+ (s, c, req, done) -> s.listReplicationPeers(c, req, done),
+ (resp) -> resp.getPeerDescList().stream()
+ .map(ReplicationSerDeHelper::toReplicationPeerDescription)
+ .collect(Collectors.toList()))).call();
+ }
+
+ @Override
+ public CompletableFuture<List<TableCFs>> listReplicatedTableCFs() {
+ CompletableFuture<List<TableCFs>> future = new CompletableFuture<List<TableCFs>>();
+ listTables().whenComplete(
+ (tables, error) -> {
+ if (!completeExceptionally(future, error)) {
+ List<TableCFs> replicatedTableCFs = new ArrayList<>();
+ Arrays.asList(tables).forEach(
+ table -> {
+ Map<String, Integer> cfs = new HashMap<>();
+ Arrays.asList(table.getColumnFamilies()).stream()
+ .filter(column -> column.getScope() != HConstants.REPLICATION_SCOPE_LOCAL)
+ .forEach(column -> {
+ cfs.put(column.getNameAsString(), column.getScope());
+ });
+ if (!cfs.isEmpty()) {
+ replicatedTableCFs.add(new TableCFs(table.getTableName(), cfs));
+ }
+ });
+ future.complete(replicatedTableCFs);
}
});
return future;
@@ -1470,4 +1661,12 @@ public class AsyncHBaseAdmin implements AsyncAdmin {
future.completeExceptionally(error);
return future;
}
+
+ private <T> boolean completeExceptionally(CompletableFuture<T> future, Throwable error) {
+ if (error != null) {
+ future.completeExceptionally(error);
+ return true;
+ }
+ return false;
+ }
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/40cc666a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HBaseAdmin.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HBaseAdmin.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HBaseAdmin.java
index cadd6cc..8063070 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HBaseAdmin.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/HBaseAdmin.java
@@ -3869,31 +3869,7 @@ public class HBaseAdmin implements Admin {
throw new ReplicationException("tableCfs is null");
}
ReplicationPeerConfig peerConfig = getReplicationPeerConfig(id);
- Map<TableName, List<String>> preTableCfs = peerConfig.getTableCFsMap();
- if (preTableCfs == null) {
- peerConfig.setTableCFsMap(tableCfs);
- } else {
- for (Map.Entry<TableName, ? extends Collection<String>> entry : tableCfs.entrySet()) {
- TableName table = entry.getKey();
- Collection<String> appendCfs = entry.getValue();
- if (preTableCfs.containsKey(table)) {
- List<String> cfs = preTableCfs.get(table);
- if (cfs == null || appendCfs == null || appendCfs.isEmpty()) {
- preTableCfs.put(table, null);
- } else {
- Set<String> cfSet = new HashSet<String>(cfs);
- cfSet.addAll(appendCfs);
- preTableCfs.put(table, Lists.newArrayList(cfSet));
- }
- } else {
- if (appendCfs == null || appendCfs.isEmpty()) {
- preTableCfs.put(table, null);
- } else {
- preTableCfs.put(table, Lists.newArrayList(appendCfs));
- }
- }
- }
- }
+ ReplicationSerDeHelper.appendTableCFsToReplicationPeerConfig(tableCfs, peerConfig);
updateReplicationPeerConfig(id, peerConfig);
}
@@ -3905,37 +3881,7 @@ public class HBaseAdmin implements Admin {
throw new ReplicationException("tableCfs is null");
}
ReplicationPeerConfig peerConfig = getReplicationPeerConfig(id);
- Map<TableName, List<String>> preTableCfs = peerConfig.getTableCFsMap();
- if (preTableCfs == null) {
- throw new ReplicationException("Table-Cfs for peer" + id + " is null");
- }
- for (Map.Entry<TableName, ? extends Collection<String>> entry : tableCfs.entrySet()) {
-
- TableName table = entry.getKey();
- Collection<String> removeCfs = entry.getValue();
- if (preTableCfs.containsKey(table)) {
- List<String> cfs = preTableCfs.get(table);
- if (cfs == null && (removeCfs == null || removeCfs.isEmpty())) {
- preTableCfs.remove(table);
- } else if (cfs != null && (removeCfs != null && !removeCfs.isEmpty())) {
- Set<String> cfSet = new HashSet<String>(cfs);
- cfSet.removeAll(removeCfs);
- if (cfSet.isEmpty()) {
- preTableCfs.remove(table);
- } else {
- preTableCfs.put(table, Lists.newArrayList(cfSet));
- }
- } else if (cfs == null && (removeCfs != null && !removeCfs.isEmpty())) {
- throw new ReplicationException("Cannot remove cf of table: " + table
- + " which doesn't specify cfs from table-cfs config in peer: " + id);
- } else if (cfs != null && (removeCfs == null || removeCfs.isEmpty())) {
- throw new ReplicationException("Cannot remove table: " + table
- + " which has specified cfs from table-cfs config in peer: " + id);
- }
- } else {
- throw new ReplicationException("No table: " + table + " in table-cfs config of peer: " + id);
- }
- }
+ ReplicationSerDeHelper.removeTableCFsFromReplicationPeerConfig(tableCfs, peerConfig, id);
updateReplicationPeerConfig(id, peerConfig);
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/40cc666a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationSerDeHelper.java
----------------------------------------------------------------------
diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationSerDeHelper.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationSerDeHelper.java
index 2d5539c..f561f4a 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationSerDeHelper.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/client/replication/ReplicationSerDeHelper.java
@@ -30,11 +30,14 @@ import org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
import org.apache.hadoop.hbase.shaded.protobuf.generated.ReplicationProtos;
+import org.apache.hadoop.hbase.replication.ReplicationException;
import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Strings;
+import com.google.common.collect.Lists;
+
import java.io.IOException;
import java.util.Collection;
import java.util.HashSet;
@@ -367,4 +370,68 @@ public final class ReplicationSerDeHelper {
builder.setConfig(convert(desc.getPeerConfig()));
return builder.build();
}
+
+ public static void appendTableCFsToReplicationPeerConfig(
+ Map<TableName, ? extends Collection<String>> tableCfs, ReplicationPeerConfig peerConfig) {
+ Map<TableName, List<String>> preTableCfs = peerConfig.getTableCFsMap();
+ if (preTableCfs == null) {
+ peerConfig.setTableCFsMap(tableCfs);
+ } else {
+ for (Map.Entry<TableName, ? extends Collection<String>> entry : tableCfs.entrySet()) {
+ TableName table = entry.getKey();
+ Collection<String> appendCfs = entry.getValue();
+ if (preTableCfs.containsKey(table)) {
+ List<String> cfs = preTableCfs.get(table);
+ if (cfs == null || appendCfs == null || appendCfs.isEmpty()) {
+ preTableCfs.put(table, null);
+ } else {
+ Set<String> cfSet = new HashSet<String>(cfs);
+ cfSet.addAll(appendCfs);
+ preTableCfs.put(table, Lists.newArrayList(cfSet));
+ }
+ } else {
+ if (appendCfs == null || appendCfs.isEmpty()) {
+ preTableCfs.put(table, null);
+ } else {
+ preTableCfs.put(table, Lists.newArrayList(appendCfs));
+ }
+ }
+ }
+ }
+ }
+
+ public static void removeTableCFsFromReplicationPeerConfig(
+ Map<TableName, ? extends Collection<String>> tableCfs, ReplicationPeerConfig peerConfig,
+ String id) throws ReplicationException {
+ Map<TableName, List<String>> preTableCfs = peerConfig.getTableCFsMap();
+ if (preTableCfs == null) {
+ throw new ReplicationException("Table-Cfs for peer: " + id + " is null");
+ }
+ for (Map.Entry<TableName, ? extends Collection<String>> entry : tableCfs.entrySet()) {
+ TableName table = entry.getKey();
+ Collection<String> removeCfs = entry.getValue();
+ if (preTableCfs.containsKey(table)) {
+ List<String> cfs = preTableCfs.get(table);
+ if (cfs == null && (removeCfs == null || removeCfs.isEmpty())) {
+ preTableCfs.remove(table);
+ } else if (cfs != null && (removeCfs != null && !removeCfs.isEmpty())) {
+ Set<String> cfSet = new HashSet<String>(cfs);
+ cfSet.removeAll(removeCfs);
+ if (cfSet.isEmpty()) {
+ preTableCfs.remove(table);
+ } else {
+ preTableCfs.put(table, Lists.newArrayList(cfSet));
+ }
+ } else if (cfs == null && (removeCfs != null && !removeCfs.isEmpty())) {
+ throw new ReplicationException("Cannot remove cf of table: " + table
+ + " which doesn't specify cfs from table-cfs config in peer: " + id);
+ } else if (cfs != null && (removeCfs == null || removeCfs.isEmpty())) {
+ throw new ReplicationException("Cannot remove table: " + table
+ + " which has specified cfs from table-cfs config in peer: " + id);
+ }
+ } else {
+ throw new ReplicationException("No table: " + table + " in table-cfs config of peer: " + id);
+ }
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/hbase/blob/40cc666a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncAdminBase.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncAdminBase.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncAdminBase.java
index f0dee0a..1881d4c 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncAdminBase.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncAdminBase.java
@@ -34,7 +34,7 @@ import org.junit.BeforeClass;
*/
public abstract class TestAsyncAdminBase {
- protected static final Log LOG = LogFactory.getLog(TestAdmin1.class);
+ protected static final Log LOG = LogFactory.getLog(TestAsyncAdminBase.class);
protected final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
protected static byte[] FAMILY = Bytes.toBytes("testFamily");
protected static final byte[] FAMILY_0 = Bytes.toBytes("cf0");
http://git-wip-us.apache.org/repos/asf/hbase/blob/40cc666a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncReplicationAdminApi.java
----------------------------------------------------------------------
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncReplicationAdminApi.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncReplicationAdminApi.java
new file mode 100644
index 0000000..c850c38
--- /dev/null
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncReplicationAdminApi.java
@@ -0,0 +1,416 @@
+/**
+ * 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.client;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.CompletionException;
+
+import org.apache.hadoop.hbase.HConstants;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.replication.ReplicationException;
+import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
+import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
+import org.apache.hadoop.hbase.testclassification.ClientTests;
+import org.apache.hadoop.hbase.testclassification.MediumTests;
+import org.junit.BeforeClass;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.experimental.categories.Category;
+import org.junit.rules.TestName;
+
+/**
+ * Class to test asynchronous replication admin operations.
+ */
+@Category({MediumTests.class, ClientTests.class})
+public class TestAsyncReplicationAdminApi extends TestAsyncAdminBase {
+
+ private final String ID_ONE = "1";
+ private final String KEY_ONE = "127.0.0.1:2181:/hbase";
+ private final String ID_SECOND = "2";
+ private final String KEY_SECOND = "127.0.0.1:2181:/hbase2";
+
+ @Rule
+ public TestName name = new TestName();
+
+ @BeforeClass
+ public static void setUpBeforeClass() throws Exception {
+ TEST_UTIL.getConfiguration().setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 1);
+ TEST_UTIL.startMiniCluster();
+ ASYNC_CONN = ConnectionFactory.createAsyncConnection(TEST_UTIL.getConfiguration()).get();
+ }
+
+ @Test
+ public void testAddRemovePeer() throws Exception {
+ ReplicationPeerConfig rpc1 = new ReplicationPeerConfig();
+ rpc1.setClusterKey(KEY_ONE);
+ ReplicationPeerConfig rpc2 = new ReplicationPeerConfig();
+ rpc2.setClusterKey(KEY_SECOND);
+ // Add a valid peer
+ admin.addReplicationPeer(ID_ONE, rpc1).join();
+ // try adding the same (fails)
+ try {
+ admin.addReplicationPeer(ID_ONE, rpc1).join();
+ fail("Test case should fail as adding a same peer.");
+ } catch (CompletionException e) {
+ // OK!
+ }
+ assertEquals(1, admin.listReplicationPeers().get().size());
+ // Try to remove an inexisting peer
+ try {
+ admin.removeReplicationPeer(ID_SECOND).join();
+ fail("Test case should fail as removing a inexisting peer.");
+ } catch (CompletionException e) {
+ // OK!
+ }
+ assertEquals(1, admin.listReplicationPeers().get().size());
+ // Add a second since multi-slave is supported
+ admin.addReplicationPeer(ID_SECOND, rpc2).join();
+ assertEquals(2, admin.listReplicationPeers().get().size());
+ // Remove the first peer we added
+ admin.removeReplicationPeer(ID_ONE).join();
+ assertEquals(1, admin.listReplicationPeers().get().size());
+ admin.removeReplicationPeer(ID_SECOND).join();
+ assertEquals(0, admin.listReplicationPeers().get().size());
+ }
+
+ @Test
+ public void testPeerConfig() throws Exception {
+ ReplicationPeerConfig config = new ReplicationPeerConfig();
+ config.setClusterKey(KEY_ONE);
+ config.getConfiguration().put("key1", "value1");
+ config.getConfiguration().put("key2", "value2");
+ admin.addReplicationPeer(ID_ONE, config).join();
+
+ List<ReplicationPeerDescription> peers = admin.listReplicationPeers().get();
+ assertEquals(1, peers.size());
+ ReplicationPeerDescription peerOne = peers.get(0);
+ assertNotNull(peerOne);
+ assertEquals("value1", peerOne.getPeerConfig().getConfiguration().get("key1"));
+ assertEquals("value2", peerOne.getPeerConfig().getConfiguration().get("key2"));
+
+ admin.removeReplicationPeer(ID_ONE).join();
+ }
+
+ @Test
+ public void testEnableDisablePeer() throws Exception {
+ ReplicationPeerConfig rpc1 = new ReplicationPeerConfig();
+ rpc1.setClusterKey(KEY_ONE);
+ admin.addReplicationPeer(ID_ONE, rpc1).join();
+ List<ReplicationPeerDescription> peers = admin.listReplicationPeers().get();
+ assertEquals(1, peers.size());
+ assertTrue(peers.get(0).isEnabled());
+
+ admin.disableReplicationPeer(ID_ONE).join();
+ peers = admin.listReplicationPeers().get();
+ assertEquals(1, peers.size());
+ assertFalse(peers.get(0).isEnabled());
+ admin.removeReplicationPeer(ID_ONE).join();
+ }
+
+ @Test
+ public void testAppendPeerTableCFs() throws Exception {
+ ReplicationPeerConfig rpc1 = new ReplicationPeerConfig();
+ rpc1.setClusterKey(KEY_ONE);
+ final TableName tableName1 = TableName.valueOf(name.getMethodName() + "t1");
+ final TableName tableName2 = TableName.valueOf(name.getMethodName() + "t2");
+ final TableName tableName3 = TableName.valueOf(name.getMethodName() + "t3");
+ final TableName tableName4 = TableName.valueOf(name.getMethodName() + "t4");
+ final TableName tableName5 = TableName.valueOf(name.getMethodName() + "t5");
+ final TableName tableName6 = TableName.valueOf(name.getMethodName() + "t6");
+
+ // Add a valid peer
+ admin.addReplicationPeer(ID_ONE, rpc1).join();
+
+ Map<TableName, List<String>> tableCFs = new HashMap<>();
+
+ // append table t1 to replication
+ tableCFs.put(tableName1, null);
+ admin.appendReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ Map<TableName, List<String>> result = admin.getReplicationPeerConfig(ID_ONE).get()
+ .getTableCFsMap();
+ assertEquals(1, result.size());
+ assertEquals(true, result.containsKey(tableName1));
+ assertNull(result.get(tableName1));
+
+ // append table t2 to replication
+ tableCFs.clear();
+ tableCFs.put(tableName2, null);
+ admin.appendReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ result = admin.getReplicationPeerConfig(ID_ONE).get().getTableCFsMap();
+ assertEquals(2, result.size());
+ assertTrue("Should contain t1", result.containsKey(tableName1));
+ assertTrue("Should contain t2", result.containsKey(tableName2));
+ assertNull(result.get(tableName1));
+ assertNull(result.get(tableName2));
+
+ // append table column family: f1 of t3 to replication
+ tableCFs.clear();
+ tableCFs.put(tableName3, new ArrayList<>());
+ tableCFs.get(tableName3).add("f1");
+ admin.appendReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ result = admin.getReplicationPeerConfig(ID_ONE).get().getTableCFsMap();
+ assertEquals(3, result.size());
+ assertTrue("Should contain t1", result.containsKey(tableName1));
+ assertTrue("Should contain t2", result.containsKey(tableName2));
+ assertTrue("Should contain t3", result.containsKey(tableName3));
+ assertNull(result.get(tableName1));
+ assertNull(result.get(tableName2));
+ assertEquals(1, result.get(tableName3).size());
+ assertEquals("f1", result.get(tableName3).get(0));
+
+ // append table column family: f1,f2 of t4 to replication
+ tableCFs.clear();
+ tableCFs.put(tableName4, new ArrayList<>());
+ tableCFs.get(tableName4).add("f1");
+ tableCFs.get(tableName4).add("f2");
+ admin.appendReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ result = admin.getReplicationPeerConfig(ID_ONE).get().getTableCFsMap();
+ assertEquals(4, result.size());
+ assertTrue("Should contain t1", result.containsKey(tableName1));
+ assertTrue("Should contain t2", result.containsKey(tableName2));
+ assertTrue("Should contain t3", result.containsKey(tableName3));
+ assertTrue("Should contain t4", result.containsKey(tableName4));
+ assertNull(result.get(tableName1));
+ assertNull(result.get(tableName2));
+ assertEquals(1, result.get(tableName3).size());
+ assertEquals("f1", result.get(tableName3).get(0));
+ assertEquals(2, result.get(tableName4).size());
+ assertEquals("f1", result.get(tableName4).get(0));
+ assertEquals("f2", result.get(tableName4).get(1));
+
+ // append "table5" => [], then append "table5" => ["f1"]
+ tableCFs.clear();
+ tableCFs.put(tableName5, new ArrayList<>());
+ admin.appendReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ tableCFs.clear();
+ tableCFs.put(tableName5, new ArrayList<>());
+ tableCFs.get(tableName5).add("f1");
+ admin.appendReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ result = admin.getReplicationPeerConfig(ID_ONE).get().getTableCFsMap();
+ assertEquals(5, result.size());
+ assertTrue("Should contain t5", result.containsKey(tableName5));
+ // null means replication all cfs of tab5
+ assertNull(result.get(tableName5));
+
+ // append "table6" => ["f1"], then append "table6" => []
+ tableCFs.clear();
+ tableCFs.put(tableName6, new ArrayList<>());
+ tableCFs.get(tableName6).add("f1");
+ admin.appendReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ tableCFs.clear();
+ tableCFs.put(tableName6, new ArrayList<>());
+ admin.appendReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ result = admin.getReplicationPeerConfig(ID_ONE).get().getTableCFsMap();
+ assertEquals(6, result.size());
+ assertTrue("Should contain t6", result.containsKey(tableName6));
+ // null means replication all cfs of tab6
+ assertNull(result.get(tableName6));
+
+ admin.removeReplicationPeer(ID_ONE).join();
+ }
+
+ @Test
+ public void testRemovePeerTableCFs() throws Exception {
+ ReplicationPeerConfig rpc1 = new ReplicationPeerConfig();
+ rpc1.setClusterKey(KEY_ONE);
+ final TableName tableName1 = TableName.valueOf(name.getMethodName() + "t1");
+ final TableName tableName2 = TableName.valueOf(name.getMethodName() + "t2");
+ final TableName tableName3 = TableName.valueOf(name.getMethodName() + "t3");
+ final TableName tableName4 = TableName.valueOf(name.getMethodName() + "t4");
+ // Add a valid peer
+ admin.addReplicationPeer(ID_ONE, rpc1).join();
+ Map<TableName, List<String>> tableCFs = new HashMap<>();
+ try {
+ tableCFs.put(tableName3, null);
+ admin.removeReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ fail("Test case should fail as removing table-cfs from a peer whose table-cfs is null");
+ } catch (CompletionException e) {
+ assertTrue(e.getCause() instanceof ReplicationException);
+ }
+ assertNull(admin.getReplicationPeerConfig(ID_ONE).get().getTableCFsMap());
+
+ tableCFs.clear();
+ tableCFs.put(tableName1, null);
+ tableCFs.put(tableName2, new ArrayList<>());
+ tableCFs.get(tableName2).add("cf1");
+ admin.appendReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ try {
+ tableCFs.clear();
+ tableCFs.put(tableName3, null);
+ admin.removeReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ fail("Test case should fail as removing table-cfs from a peer whose table-cfs didn't contain t3");
+ } catch (CompletionException e) {
+ assertTrue(e.getCause() instanceof ReplicationException);
+ }
+ Map<TableName, List<String>> result = admin.getReplicationPeerConfig(ID_ONE).get()
+ .getTableCFsMap();
+ assertEquals(2, result.size());
+ assertTrue("Should contain t1", result.containsKey(tableName1));
+ assertTrue("Should contain t2", result.containsKey(tableName2));
+ assertNull(result.get(tableName1));
+ assertEquals(1, result.get(tableName2).size());
+ assertEquals("cf1", result.get(tableName2).get(0));
+
+ try {
+ tableCFs.clear();
+ tableCFs.put(tableName1, new ArrayList<>());
+ tableCFs.get(tableName1).add("cf1");
+ admin.removeReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ fail("Test case should fail, because table t1 didn't specify cfs in peer config");
+ } catch (CompletionException e) {
+ assertTrue(e.getCause() instanceof ReplicationException);
+ }
+ tableCFs.clear();
+ tableCFs.put(tableName1, null);
+ admin.removeReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ result = admin.getReplicationPeerConfig(ID_ONE).get().getTableCFsMap();
+ assertEquals(1, result.size());
+ assertEquals(1, result.get(tableName2).size());
+ assertEquals("cf1", result.get(tableName2).get(0));
+
+ try {
+ tableCFs.clear();
+ tableCFs.put(tableName2, null);
+ admin.removeReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ fail("Test case should fail, because table t2 hase specified cfs in peer config");
+ } catch (CompletionException e) {
+ assertTrue(e.getCause() instanceof ReplicationException);
+ }
+ tableCFs.clear();
+ tableCFs.put(tableName2, new ArrayList<>());
+ tableCFs.get(tableName2).add("cf1");
+ admin.removeReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ assertNull(admin.getReplicationPeerConfig(ID_ONE).get().getTableCFsMap());
+
+ tableCFs.clear();
+ tableCFs.put(tableName4, new ArrayList<>());
+ admin.appendReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ admin.removeReplicationPeerTableCFs(ID_ONE, tableCFs).join();
+ assertNull(admin.getReplicationPeerConfig(ID_ONE).get().getTableCFsMap());
+
+ admin.removeReplicationPeer(ID_ONE);
+ }
+
+ @Test
+ public void testSetPeerNamespaces() throws Exception {
+ String ns1 = "ns1";
+ String ns2 = "ns2";
+
+ ReplicationPeerConfig rpc = new ReplicationPeerConfig();
+ rpc.setClusterKey(KEY_ONE);
+ admin.addReplicationPeer(ID_ONE, rpc).join();
+
+ // add ns1 and ns2 to peer config
+ rpc = admin.getReplicationPeerConfig(ID_ONE).get();
+ Set<String> namespaces = new HashSet<>();
+ namespaces.add(ns1);
+ namespaces.add(ns2);
+ rpc.setNamespaces(namespaces);
+ admin.updateReplicationPeerConfig(ID_ONE, rpc).join();
+ namespaces = admin.getReplicationPeerConfig(ID_ONE).get().getNamespaces();
+ assertEquals(2, namespaces.size());
+ assertTrue(namespaces.contains(ns1));
+ assertTrue(namespaces.contains(ns2));
+
+ // update peer config only contains ns1
+ rpc = admin.getReplicationPeerConfig(ID_ONE).get();
+ namespaces.clear();
+ namespaces.add(ns1);
+ rpc.setNamespaces(namespaces);
+ admin.updateReplicationPeerConfig(ID_ONE, rpc).join();
+ namespaces = admin.getReplicationPeerConfig(ID_ONE).get().getNamespaces();
+ assertEquals(1, namespaces.size());
+ assertTrue(namespaces.contains(ns1));
+
+ admin.removeReplicationPeer(ID_ONE).join();
+ }
+
+ @Test
+ public void testNamespacesAndTableCfsConfigConflict() throws Exception {
+ String ns1 = "ns1";
+ String ns2 = "ns2";
+ final TableName tableName1 = TableName.valueOf(ns1 + ":" + name.getMethodName());
+ final TableName tableName2 = TableName.valueOf(ns2 + ":" + name.getMethodName() + "2");
+
+ ReplicationPeerConfig rpc = new ReplicationPeerConfig();
+ rpc.setClusterKey(KEY_ONE);
+ admin.addReplicationPeer(ID_ONE, rpc).join();
+
+ rpc = admin.getReplicationPeerConfig(ID_ONE).get();
+ Set<String> namespaces = new HashSet<String>();
+ namespaces.add(ns1);
+ rpc.setNamespaces(namespaces);
+ admin.updateReplicationPeerConfig(ID_ONE, rpc).get();
+ rpc = admin.getReplicationPeerConfig(ID_ONE).get();
+ Map<TableName, List<String>> tableCfs = new HashMap<>();
+ tableCfs.put(tableName1, new ArrayList<>());
+ rpc.setTableCFsMap(tableCfs);
+ try {
+ admin.updateReplicationPeerConfig(ID_ONE, rpc).join();
+ fail("Test case should fail, because table " + tableName1 + " conflict with namespace " + ns1);
+ } catch (CompletionException e) {
+ // OK
+ }
+
+ rpc = admin.getReplicationPeerConfig(ID_ONE).get();
+ tableCfs.clear();
+ tableCfs.put(tableName2, new ArrayList<>());
+ rpc.setTableCFsMap(tableCfs);
+ admin.updateReplicationPeerConfig(ID_ONE, rpc).get();
+ rpc = admin.getReplicationPeerConfig(ID_ONE).get();
+ namespaces.clear();
+ namespaces.add(ns2);
+ rpc.setNamespaces(namespaces);
+ try {
+ admin.updateReplicationPeerConfig(ID_ONE, rpc).join();
+ fail("Test case should fail, because namespace " + ns2 + " conflict with table " + tableName2);
+ } catch (CompletionException e) {
+ // OK
+ }
+
+ admin.removeReplicationPeer(ID_ONE).join();
+ }
+
+ @Test
+ public void testPeerBandwidth() throws Exception {
+ ReplicationPeerConfig rpc = new ReplicationPeerConfig();
+ rpc.setClusterKey(KEY_ONE);
+
+ admin.addReplicationPeer(ID_ONE, rpc).join();
+ rpc = admin.getReplicationPeerConfig(ID_ONE).get();
+ assertEquals(0, rpc.getBandwidth());
+
+ rpc.setBandwidth(2097152);
+ admin.updateReplicationPeerConfig(ID_ONE, rpc).join();
+ assertEquals(2097152, admin.getReplicationPeerConfig(ID_ONE).join().getBandwidth());
+
+ admin.removeReplicationPeer(ID_ONE).join();
+ }
+}