From 581a7b7f91e37f5381ad83de5bc7815e8207589e Mon Sep 17 00:00:00 2001
From: "yoshida.shinya" <yoshida.shinya@linecorp.com>
Date: Fri, 13 Aug 2021 11:30:58 +0900
Subject: [PATCH] HBASE-26196 Support configuration override for remote cluster
 of HFileOutputFormat locality sensitive

---
 .../hbase/mapreduce/HFileOutputFormat2.java   | 26 ++++++++++++++++---
 .../mapreduce/TestHFileOutputFormat2.java     |  9 +++++++
 2 files changed, 32 insertions(+), 3 deletions(-)

diff --git a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/HFileOutputFormat2.java b/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/HFileOutputFormat2.java
index dd4266103f0a..43bb594ec190 100644
--- a/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/HFileOutputFormat2.java
+++ b/hbase-mapreduce/src/main/java/org/apache/hadoop/hbase/mapreduce/HFileOutputFormat2.java
@@ -32,6 +32,7 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.Map;
+import java.util.Map.Entry;
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.TreeSet;
@@ -163,12 +164,14 @@ protected static byte[] combineTableNameSuffix(byte[] tableName, byte[] suffix)
   static final String MULTI_TABLE_HFILEOUTPUTFORMAT_CONF_KEY =
           "hbase.mapreduce.use.multi.table.hfileoutputformat";
 
+  public static final String REMOTE_CLUSTER_CONF_PREFIX =
+    "hbase.hfileoutputformat.remote.cluster.";
   public static final String REMOTE_CLUSTER_ZOOKEEPER_QUORUM_CONF_KEY =
-    "hbase.hfileoutputformat.remote.cluster.zookeeper.quorum";
+    REMOTE_CLUSTER_CONF_PREFIX + "zookeeper.quorum";
   public static final String REMOTE_CLUSTER_ZOOKEEPER_CLIENT_PORT_CONF_KEY =
-    "hbase.hfileoutputformat.remote.cluster.zookeeper." + HConstants.CLIENT_PORT_STR;
+    REMOTE_CLUSTER_CONF_PREFIX + "zookeeper." + HConstants.CLIENT_PORT_STR;
   public static final String REMOTE_CLUSTER_ZOOKEEPER_ZNODE_PARENT_CONF_KEY =
-    "hbase.hfileoutputformat.remote.cluster." + HConstants.ZOOKEEPER_ZNODE_PARENT;
+    REMOTE_CLUSTER_CONF_PREFIX + HConstants.ZOOKEEPER_ZNODE_PARENT;
 
   public static final String STORAGE_POLICY_PROPERTY = HStore.BLOCK_STORAGE_POLICY_KEY;
   public static final String STORAGE_POLICY_PROPERTY_CF_PREFIX = STORAGE_POLICY_PROPERTY + ".";
@@ -361,6 +364,23 @@ private Configuration createRemoteClusterConf(Configuration conf) {
           newConf.set(HConstants.ZOOKEEPER_ZNODE_PARENT, parent);
         }
 
+        for (Entry<String, String> entry : conf) {
+          String key = entry.getKey();
+          if (REMOTE_CLUSTER_ZOOKEEPER_QUORUM_CONF_KEY.equals(key) ||
+              REMOTE_CLUSTER_ZOOKEEPER_CLIENT_PORT_CONF_KEY.equals(key) ||
+              REMOTE_CLUSTER_ZOOKEEPER_ZNODE_PARENT_CONF_KEY.equals(key)) {
+            // Handled them above
+            continue;
+          }
+
+          if (entry.getKey().startsWith(REMOTE_CLUSTER_CONF_PREFIX)) {
+            String originalKey = entry.getKey().substring(REMOTE_CLUSTER_CONF_PREFIX.length());
+            if (!originalKey.isEmpty()) {
+              newConf.set(originalKey, entry.getValue());
+            }
+          }
+        }
+
         return newConf;
       }
 
diff --git a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestHFileOutputFormat2.java b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestHFileOutputFormat2.java
index 12f1a624c5c1..f45eb4f4fc9a 100644
--- a/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestHFileOutputFormat2.java
+++ b/hbase-mapreduce/src/test/java/org/apache/hadoop/hbase/mapreduce/TestHFileOutputFormat2.java
@@ -102,6 +102,7 @@
 import org.apache.hadoop.hbase.regionserver.HStore;
 import org.apache.hadoop.hbase.regionserver.TestHRegionFileSystem;
 import org.apache.hadoop.hbase.regionserver.TimeRangeTracker;
+import org.apache.hadoop.hbase.security.SecurityConstants;
 import org.apache.hadoop.hbase.security.User;
 import org.apache.hadoop.hbase.testclassification.LargeTests;
 import org.apache.hadoop.hbase.testclassification.VerySlowMapReduceTests;
@@ -1693,6 +1694,11 @@ public void testMRIncrementalLoadWithLocalityMultiCluster() throws Exception {
       assertEquals(confB.get(HConstants.ZOOKEEPER_ZNODE_PARENT),
         jobConf.get(HFileOutputFormat2.REMOTE_CLUSTER_ZOOKEEPER_ZNODE_PARENT_CONF_KEY));
 
+      String bSpecificConfigKey = "my.override.config.for.b";
+      String bSpecificConfigValue = "b-specific-value";
+      jobConf.set(HFileOutputFormat2.REMOTE_CLUSTER_CONF_PREFIX + bSpecificConfigKey,
+        bSpecificConfigValue);
+
       FileOutputFormat.setOutputPath(job, testDir);
 
       assertFalse(util.getTestFileSystem().exists(testDir));
@@ -1710,6 +1716,9 @@ public void testMRIncrementalLoadWithLocalityMultiCluster() throws Exception {
           config.get(HConstants.ZOOKEEPER_CLIENT_PORT));
         assertEquals(confB.get(HConstants.ZOOKEEPER_ZNODE_PARENT),
           config.get(HConstants.ZOOKEEPER_ZNODE_PARENT));
+
+        assertEquals(bSpecificConfigValue,
+          config.get(bSpecificConfigKey));
       }
     } finally {
       utilB.deleteTable(tableName);