flink-connector-hbase 添加配置项hbase.conf.dir

dongkelun · Jul 13, 2023 · c41cfed · c41cfed
1 parent fc679b6
commit c41cfed
Show file tree

Hide file tree

Showing 3 changed files with 12 additions and 0 deletions.
diff --git a/...-hbase-2.2/src/main/java/org/apache/flink/connector/hbase2/HBase2DynamicTableFactory.java b/...-hbase-2.2/src/main/java/org/apache/flink/connector/hbase2/HBase2DynamicTableFactory.java
@@ -39,6 +39,7 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.apache.flink.connector.hbase.table.HBaseConnectorOptions.HBASE_CONF_DIR;
 import static org.apache.flink.connector.hbase.table.HBaseConnectorOptions.LOOKUP_ASYNC;
 import static org.apache.flink.connector.hbase.table.HBaseConnectorOptions.LOOKUP_CACHE_MAX_ROWS;
 import static org.apache.flink.connector.hbase.table.HBaseConnectorOptions.LOOKUP_CACHE_TTL;
@@ -131,6 +132,7 @@ public Set<ConfigOption<?>> optionalOptions() {
         set.add(LOOKUP_CACHE_MAX_ROWS);
         set.add(LOOKUP_CACHE_TTL);
         set.add(LOOKUP_MAX_RETRIES);
+        set.add(HBASE_CONF_DIR);
         return set;
     }
 

diff --git a/...base-base/src/main/java/org/apache/flink/connector/hbase/table/HBaseConnectorOptions.java b/...base-base/src/main/java/org/apache/flink/connector/hbase/table/HBaseConnectorOptions.java
@@ -51,6 +51,12 @@ public class HBaseConnectorOptions {
                     .defaultValue("/hbase")
                     .withDescription("The root dir in Zookeeper for HBase cluster.");
 
+    public static final ConfigOption<String> HBASE_CONF_DIR =
+            ConfigOptions.key("hbase.conf.dir")
+                    .stringType()
+                    .defaultValue("/etc/hbase/conf")
+                    .withDescription("Configuration folder path for hbase.");
+
     public static final ConfigOption<String> NULL_STRING_LITERAL =
             ConfigOptions.key("null-string-literal")
                     .stringType()

diff --git a/...-base/src/main/java/org/apache/flink/connector/hbase/table/HBaseConnectorOptionsUtil.java b/...-base/src/main/java/org/apache/flink/connector/hbase/table/HBaseConnectorOptionsUtil.java
@@ -32,6 +32,7 @@
 import java.util.Map;
 import java.util.Properties;
 
+import static org.apache.flink.connector.hbase.table.HBaseConnectorOptions.HBASE_CONF_DIR;
 import static org.apache.flink.connector.hbase.table.HBaseConnectorOptions.LOOKUP_CACHE_MAX_ROWS;
 import static org.apache.flink.connector.hbase.table.HBaseConnectorOptions.SINK_BUFFER_FLUSH_INTERVAL;
 import static org.apache.flink.connector.hbase.table.HBaseConnectorOptions.SINK_BUFFER_FLUSH_MAX_ROWS;
@@ -118,6 +119,9 @@ public static Configuration getHBaseConfiguration(ReadableConfig tableOptions) {
                 getHBaseClientProperties(
                         ((org.apache.flink.configuration.Configuration) tableOptions).toMap());
         properties.forEach((k, v) -> hbaseClientConf.set(k.toString(), v.toString()));
+        hbaseClientConf.addResource(
+                new org.apache.hadoop.fs.Path(
+                        tableOptions.get(HBASE_CONF_DIR) + "/hbase-site.xml"));
         return hbaseClientConf;
     }