CDPD-71667: Add CDPD-34408 and COD-1499 on top of the feature branch (apache#209)

* CDPD-71667: Revert "HBASE-24764: Add support of adding default peer configs via hbase-site.xml for all replication peers. (apache#2284)"

This reverts commit 7df1b92.

Change-Id: I2b4570a3ab1bca6c573c669a54ef67008438bd95

* CDPD-71667: Revert "HBASE-24743 Reject to add a peer which replicate to itself earlier (apache#2124)"

This reverts commit 70ab0dc.

Change-Id: I6d508ff3aacf2d48e00f903afe7bca21d34f3247

---------

Co-authored-by: Wellington Ramos Chevreuil <wchevreuil@apache.org>

Author: wchevreuil

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

@@ -803,7 +803,7 @@ private void initializeZKBasedSystemTrackers()
     this.splitOrMergeStateStore = new SplitOrMergeStateStore(masterRegion, zooKeeper, conf);
 
     this.replicationPeerManager =
-      ReplicationPeerManager.create(fileSystemManager.getFileSystem(), zooKeeper, conf, clusterId);
+      ReplicationPeerManager.create(fileSystemManager.getFileSystem(), zooKeeper, conf);
     this.configurationManager.registerObserver(replicationPeerManager);
     this.replicationPeerModificationStateStore =
       new ReplicationPeerModificationStateStore(masterRegion);

hbase-server/src/main/java/org/apache/hadoop/hbase/master/replication/ReplicationPeerManager.java

@@ -31,6 +31,7 @@
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.hbase.DoNotRetryIOException;
 import org.apache.hadoop.hbase.HBaseConfiguration;
 import org.apache.hadoop.hbase.ReplicationPeerNotFoundException;
@@ -50,6 +51,7 @@
 import org.apache.hadoop.hbase.replication.ReplicationQueueStorage;
 import org.apache.hadoop.hbase.replication.ReplicationStorageFactory;
 import org.apache.hadoop.hbase.replication.ReplicationUtils;
+import org.apache.hadoop.hbase.replication.regionserver.HBaseInterClusterReplicationEndpoint;
 import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
 import org.apache.hadoop.hbase.zookeeper.ZKConfig;
 import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
@@ -73,25 +75,18 @@ public class ReplicationPeerManager implements ConfigurationObserver {
 
   private final ConcurrentMap<String, ReplicationPeerDescription> peers;
 
-  private final String clusterId;
-
-  private volatile Configuration conf;
-
   // for dynamic recreating ReplicationPeerStorage.
   private final FileSystem fs;
 
   private final ZKWatcher zk;
 
-  ReplicationPeerManager(FileSystem fs, ZKWatcher zk, ReplicationPeerStorage peerStorage,
-    ReplicationQueueStorage queueStorage, ConcurrentMap<String, ReplicationPeerDescription> peers,
-    Configuration conf, String clusterId) {
-    this.fs = fs;
-    this.zk = zk;
+  ReplicationPeerManager(FileSystem fs, ZKWatcher zk, ReplicationPeerStorage peerStorage, ReplicationQueueStorage queueStorage,
+      ConcurrentMap<String, ReplicationPeerDescription> peers) {
     this.peerStorage = peerStorage;
     this.queueStorage = queueStorage;
     this.peers = peers;
-    this.conf = conf;
-    this.clusterId = clusterId;
+    this.fs = fs;
+    this.zk = zk;
   }
 
   private void checkQueuesDeleted(String peerId)
@@ -181,12 +176,11 @@ ReplicationPeerDescription preUpdatePeerConfig(String peerId, ReplicationPeerCon
   }
 
   public void addPeer(String peerId, ReplicationPeerConfig peerConfig, boolean enabled)
-    throws ReplicationException {
+      throws ReplicationException {
     if (peers.containsKey(peerId)) {
       // this should be a retry, just return
       return;
     }
-    peerConfig = ReplicationPeerConfigUtil.updateReplicationBasePeerConfigs(conf, peerConfig);
     ReplicationPeerConfig copiedPeerConfig = ReplicationPeerConfig.newBuilder(peerConfig).build();
     peerStorage.addPeer(peerId, copiedPeerConfig, enabled);
     peers.put(peerId, new ReplicationPeerDescription(peerId, enabled, copiedPeerConfig));
@@ -281,25 +275,25 @@ void removeAllQueuesAndHFileRefs(String peerId) throws ReplicationException {
 
   private void checkPeerConfig(ReplicationPeerConfig peerConfig) throws DoNotRetryIOException {
     String replicationEndpointImpl = peerConfig.getReplicationEndpointImpl();
-    ReplicationEndpoint endpoint = null;
+    boolean checkClusterKey = true;
     if (!StringUtils.isBlank(replicationEndpointImpl)) {
+      // try creating a instance
+      ReplicationEndpoint endpoint;
       try {
-        // try creating a instance
-        endpoint = Class.forName(replicationEndpointImpl).asSubclass(ReplicationEndpoint.class)
-          .getDeclaredConstructor().newInstance();
+        endpoint = Class.forName(replicationEndpointImpl)
+          .asSubclass(ReplicationEndpoint.class).getDeclaredConstructor().newInstance();
       } catch (Throwable e) {
         throw new DoNotRetryIOException(
           "Can not instantiate configured replication endpoint class=" + replicationEndpointImpl,
           e);
       }
+      // do not check cluster key if we are not HBaseInterClusterReplicationEndpoint
+      if (!(endpoint instanceof HBaseInterClusterReplicationEndpoint)) {
+        checkClusterKey = false;
+      }
     }
-    // Endpoints implementing HBaseReplicationEndpoint need to check cluster key
-    if (endpoint == null || endpoint instanceof HBaseReplicationEndpoint) {
+    if (checkClusterKey) {
       checkClusterKey(peerConfig.getClusterKey());
-      // Check if endpoint can replicate to the same cluster
-      if (endpoint == null || !endpoint.canReplicateToSameCluster()) {
-        checkSameClusterKey(peerConfig.getClusterKey());
-      }
     }
 
     if (peerConfig.replicateAllUserTables()) {
@@ -396,25 +390,6 @@ private void checkClusterKey(String clusterKey) throws DoNotRetryIOException {
     }
   }
 
-  private void checkSameClusterKey(String clusterKey) throws DoNotRetryIOException {
-    String peerClusterId = "";
-    try {
-      // Create the peer cluster config for get peer cluster id
-      Configuration peerConf = HBaseConfiguration.createClusterConf(conf, clusterKey);
-      try (ZKWatcher zkWatcher = new ZKWatcher(peerConf, this + "check-peer-cluster-id", null)) {
-        peerClusterId = ZKClusterId.readClusterIdZNode(zkWatcher);
-      }
-    } catch (IOException | KeeperException e) {
-      throw new DoNotRetryIOException("Can't get peerClusterId for clusterKey=" + clusterKey, e);
-    }
-    // In rare case, zookeeper setting may be messed up. That leads to the incorrect
-    // peerClusterId value, which is the same as the source clusterId
-    if (clusterId.equals(peerClusterId)) {
-      throw new DoNotRetryIOException("Invalid cluster key: " + clusterKey
-        + ", should not replicate to itself for HBaseInterClusterReplicationEndpoint");
-    }
-  }
-
   public List<String> getSerialPeerIdsBelongsTo(TableName tableName) {
     return peers.values().stream().filter(p -> p.getPeerConfig().isSerial())
       .filter(p -> p.getPeerConfig().needToReplicate(tableName)).map(p -> p.getPeerId())
@@ -425,21 +400,18 @@ public ReplicationQueueStorage getQueueStorage() {
     return queueStorage;
   }
 
-  public static ReplicationPeerManager create(FileSystem fs, ZKWatcher zk, Configuration conf,
-    String clusterId) throws ReplicationException {
+  public static ReplicationPeerManager create(FileSystem fs, ZKWatcher zk, Configuration conf)
+      throws ReplicationException {
     ReplicationPeerStorage peerStorage =
       ReplicationStorageFactory.getReplicationPeerStorage(fs, zk, conf);
     ConcurrentMap<String, ReplicationPeerDescription> peers = new ConcurrentHashMap<>();
     for (String peerId : peerStorage.listPeerIds()) {
       ReplicationPeerConfig peerConfig = peerStorage.getPeerConfig(peerId);
-
-      peerConfig = ReplicationPeerConfigUtil.updateReplicationBasePeerConfigs(conf, peerConfig);
-      peerStorage.updatePeerConfig(peerId, peerConfig);
       boolean enabled = peerStorage.isPeerEnabled(peerId);
       peers.put(peerId, new ReplicationPeerDescription(peerId, enabled, peerConfig));
     }
     return new ReplicationPeerManager(fs, zk, peerStorage,
-      ReplicationStorageFactory.getReplicationQueueStorage(zk, conf), peers, conf, clusterId);
+      ReplicationStorageFactory.getReplicationQueueStorage(zk, conf), peers);
   }
 
   /**
@@ -455,7 +427,6 @@ private boolean isStringEquals(String s1, String s2) {
 
   @Override
   public void onConfigurationChange(Configuration conf) {
-    this.conf = conf;
     this.peerStorage = ReplicationStorageFactory.getReplicationPeerStorage(fs, zk, conf);
   }
 }

hbase-server/src/main/java/org/apache/hadoop/hbase/replication/regionserver/ReplicationSource.java

@@ -584,6 +584,18 @@ private void initialize() {
       return;
     }
 
+    // In rare case, zookeeper setting may be messed up. That leads to the incorrect
+    // peerClusterId value, which is the same as the source clusterId
+    if (clusterId.equals(peerClusterId) && !replicationEndpoint.canReplicateToSameCluster()) {
+      this.terminate("ClusterId " + clusterId + " is replicating to itself: peerClusterId "
+          + peerClusterId + " which is not allowed by ReplicationEndpoint:"
+          + replicationEndpoint.getClass().getName(), null, false);
+      this.manager.removeSource(this);
+      return;
+    }
+    LOG.info("{} Source: {}, is now replicating from cluster: {}; to peer cluster: {};",
+      logPeerId(), this.replicationQueueInfo.getQueueId(), clusterId, peerClusterId);
+
     LOG.info("{} queueId={} (queues={}) is replicating from cluster={} to cluster={}", logPeerId(),
       this.replicationQueueInfo.getQueueId(), logQueue.getNumQueues(), clusterId, peerClusterId);
     initializeWALEntryFilter(peerClusterId);

hbase-server/src/test/java/org/apache/hadoop/hbase/client/TestAsyncReplicationAdminApi.java

@@ -74,9 +74,9 @@ public class TestAsyncReplicationAdminApi extends TestAsyncAdminBase {
     HBaseClassTestRule.forClass(TestAsyncReplicationAdminApi.class);
 
   private final String ID_ONE = "1";
-  private static String KEY_ONE;
+  private final String KEY_ONE = "127.0.0.1:2181:/hbase";
   private final String ID_TWO = "2";
-  private static String KEY_TWO;
+  private final String KEY_TWO = "127.0.0.1:2181:/hbase2";
 
   @BeforeClass
   public static void setUpBeforeClass() throws Exception {
@@ -85,8 +85,6 @@ public static void setUpBeforeClass() throws Exception {
     TEST_UTIL.getConfiguration().setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 2);
     TEST_UTIL.getConfiguration().setInt(START_LOG_ERRORS_AFTER_COUNT_KEY, 0);
     TEST_UTIL.startMiniCluster();
-    KEY_ONE = TEST_UTIL.getClusterKey() + "-test1";
-    KEY_TWO = TEST_UTIL.getClusterKey() + "-test2";
     ASYNC_CONN = ConnectionFactory.createAsyncConnection(TEST_UTIL.getConfiguration()).get();
   }
 

hbase-server/src/test/java/org/apache/hadoop/hbase/client/replication/TestReplicationAdmin.java

@@ -77,9 +77,9 @@ public class TestReplicationAdmin {
   private final static HBaseTestingUtility TEST_UTIL = new HBaseTestingUtility();
 
   private final String ID_ONE = "1";
-  private static String KEY_ONE;
+  private final String KEY_ONE = "127.0.0.1:2181:/hbase";
   private final String ID_SECOND = "2";
-  private static String KEY_SECOND;
+  private final String KEY_SECOND = "127.0.0.1:2181:/hbase2";
 
   private static ReplicationAdmin admin;
   private static Admin hbaseAdmin;
@@ -96,8 +96,6 @@ public static void setUpBeforeClass() throws Exception {
     TEST_UTIL.startMiniCluster();
     admin = new ReplicationAdmin(TEST_UTIL.getConfiguration());
     hbaseAdmin = TEST_UTIL.getAdmin();
-    KEY_ONE = TEST_UTIL.getClusterKey() + "-test1";
-    KEY_SECOND = TEST_UTIL.getClusterKey() + "-test2";
   }
 
   @AfterClass

hbase-server/src/test/java/org/apache/hadoop/hbase/replication/SerialReplicationTestBase.java

@@ -114,11 +114,6 @@ protected void doStart() {
     protected void doStop() {
       notifyStopped();
     }
-
-    @Override
-    public boolean canReplicateToSameCluster() {
-      return true;
-    }
   }
 
   @BeforeClass