org.apache.hadoop.hbase.replication.ReplicationPeerConfig#setReplicateAllUserTables ( )源码实例Demo

@Test
public void testEnableReplicationForExplicitSetTableCfs() throws Exception {
  TableName tableName2 = TableName.valueOf(tableName.getNameAsString() + "2");
  // Only create table in source cluster
  createTableWithDefaultConf(tableName);
  createTableWithDefaultConf(tableName2);
  assertFalse("Table should not exists in the peer cluster",
    admin2.tableExists(tableName).get());
  assertFalse("Table should not exists in the peer cluster",
    admin2.tableExists(tableName2).get());

  Map<TableName, ? extends Collection<String>> tableCfs = new HashMap<>();
  tableCfs.put(tableName, null);
  ReplicationPeerConfig rpc = admin.getReplicationPeerConfig(ID_SECOND).get();
  rpc.setReplicateAllUserTables(false);
  rpc.setTableCFsMap(tableCfs);
  try {
    // Only add tableName to replication peer config
    admin.updateReplicationPeerConfig(ID_SECOND, rpc).join();
    admin.enableTableReplication(tableName2).join();
    assertFalse("Table should not be created if user has set table cfs explicitly for the "
        + "peer and this is not part of that collection", admin2.tableExists(tableName2).get());

    // Add tableName2 to replication peer config, too
    tableCfs.put(tableName2, null);
    rpc.setTableCFsMap(tableCfs);
    admin.updateReplicationPeerConfig(ID_SECOND, rpc).join();
    admin.enableTableReplication(tableName2).join();
    assertTrue(
      "Table should be created if user has explicitly added table into table cfs collection",
      admin2.tableExists(tableName2).get());
  } finally {
    rpc.setTableCFsMap(null);
    rpc.setReplicateAllUserTables(true);
    admin.updateReplicationPeerConfig(ID_SECOND, rpc).join();
  }
}
 

@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();
  rpc.setReplicateAllUserTables(false);
  admin.updateReplicationPeerConfig(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 = new HashSet<>();
  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();
}