[Feature] [Resource] Add resource hdfs ha (#3056)

DataLinkDC · Jan 24, 2024 · 7910711 · 7910711
1 parent 1400d08
commit 7910711
Show file tree

Hide file tree

Showing 8 changed files with 401 additions and 349 deletions.
diff --git a/dinky-admin/src/main/java/org/dinky/init/SystemInit.java b/dinky-admin/src/main/java/org/dinky/init/SystemInit.java
@@ -124,6 +124,8 @@ private void initResources() {
                         systemConfiguration.getResourcesOssEndpoint(),
                         systemConfiguration.getResourcesHdfsUser(),
                         systemConfiguration.getResourcesHdfsDefaultFS(),
+                        systemConfiguration.getResourcesHdfsCoreSite(),
+                        systemConfiguration.getResourcesHdfsHdfsSite(),
                         systemConfiguration.getResourcesOssAccessKey(),
                         systemConfiguration.getResourcesOssRegion(),
                         systemConfiguration.getResourcesPathStyleAccess())

diff --git a/dinky-admin/src/main/java/org/dinky/service/resource/BaseResourceManager.java b/dinky-admin/src/main/java/org/dinky/service/resource/BaseResourceManager.java
@@ -31,11 +31,15 @@
 
 import java.io.File;
 import java.io.InputStream;
+import java.nio.charset.Charset;
 
 import org.springframework.web.multipart.MultipartFile;
 
 import cn.hutool.core.io.FileUtil;
+import cn.hutool.core.io.IoUtil;
+import cn.hutool.core.lang.Opt;
 import cn.hutool.core.lang.Singleton;
+import cn.hutool.core.util.StrUtil;
 
 public interface BaseResourceManager {
     SystemConfiguration instances = SystemConfiguration.getInstances();
@@ -75,8 +79,16 @@ static void initResourceManager() {
                 break;
             case HDFS:
                 final Configuration configuration = new Configuration();
-                configuration.set(
-                        "fs.defaultFS", instances.getResourcesHdfsDefaultFS().getValue());
+                Charset charset = Charset.defaultCharset();
+                String coreSite = instances.getResourcesHdfsCoreSite().getValue();
+                Opt.ofBlankAble(coreSite).ifPresent(x -> configuration.addResource(IoUtil.toStream(x, charset), true));
+                String hdfsSite = instances.getResourcesHdfsHdfsSite().getValue();
+                Opt.ofBlankAble(hdfsSite).ifPresent(x -> configuration.addResource(IoUtil.toStream(x, charset), true));
+                if (StrUtil.isEmpty(coreSite)) {
+                    configuration.set(
+                            "fs.defaultFS",
+                            instances.getResourcesHdfsDefaultFS().getValue());
+                }
                 try {
                     FileSystem fileSystem = FileSystem.get(
                             FileSystem.getDefaultUri(configuration),

diff --git a/dinky-app/dinky-app-base/src/main/java/org/dinky/app/resource/BaseResourceManager.java b/dinky-app/dinky-app-base/src/main/java/org/dinky/app/resource/BaseResourceManager.java
@@ -30,9 +30,13 @@
 import org.apache.hadoop.fs.FileSystem;
 
 import java.io.InputStream;
+import java.nio.charset.Charset;
 
 import cn.hutool.core.io.FileUtil;
+import cn.hutool.core.io.IoUtil;
+import cn.hutool.core.lang.Opt;
 import cn.hutool.core.lang.Singleton;
+import cn.hutool.core.util.StrUtil;
 
 public interface BaseResourceManager {
     SystemConfiguration instances = SystemConfiguration.getInstances();
@@ -62,8 +66,16 @@ static void initResourceManager() {
                 break;
             case HDFS:
                 final Configuration configuration = new Configuration();
-                configuration.set(
-                        "fs.defaultFS", instances.getResourcesHdfsDefaultFS().getValue());
+                Charset charset = Charset.defaultCharset();
+                String coreSite = instances.getResourcesHdfsCoreSite().getValue();
+                Opt.ofBlankAble(coreSite).ifPresent(x -> configuration.addResource(IoUtil.toStream(x, charset), true));
+                String hdfsSite = instances.getResourcesHdfsHdfsSite().getValue();
+                Opt.ofBlankAble(hdfsSite).ifPresent(x -> configuration.addResource(IoUtil.toStream(x, charset), true));
+                if (StrUtil.isEmpty(coreSite)) {
+                    configuration.set(
+                            "fs.defaultFS",
+                            instances.getResourcesHdfsDefaultFS().getValue());
+                }
                 try {
                     FileSystem fileSystem = FileSystem.get(
                             FileSystem.getDefaultUri(configuration),

diff --git a/dinky-common/src/main/java/org/dinky/data/enums/Status.java b/dinky-common/src/main/java/org/dinky/data/enums/Status.java
@@ -402,6 +402,12 @@ public enum Status {
     SYS_RESOURCE_SETTINGS_HDFS_ROOT_USER_NOTE(173, "sys.resource.settings.hdfs.root.user.note"),
     SYS_RESOURCE_SETTINGS_HDFS_FS_DEFAULTFS(174, "sys.resource.settings.hdfs.fs.defaultFS"),
     SYS_RESOURCE_SETTINGS_HDFS_FS_DEFAULTFS_NOTE(175, "sys.resource.settings.hdfs.fs.defaultFS.note"),
+
+    SYS_RESOURCE_SETTINGS_HDFS_CORE_SITE(178, "sys.resource.settings.hdfs.core.site"),
+    SYS_RESOURCE_SETTINGS_HDFS_CORE_SITE_NOTE(179, "sys.resource.settings.hdfs.core.site.note"),
+    SYS_RESOURCE_SETTINGS_HDFS_HDFS_SITE(380, "sys.resource.settings.hdfs.hdfs.site"),
+    SYS_RESOURCE_SETTINGS_HDFS_HDFS_SITE_NOTE(381, "sys.resource.settings.hdfs.hdfs.site.note"),
+
     SYS_RESOURCE_SETTINGS_PATH_STYLE_ACCESS(176, "sys.resource.settings.oss.path.style.access"),
     SYS_RESOURCE_SETTINGS_PATH_STYLE_ACCESS_NOTE(177, "sys.resource.settings.oss.path.style.access.note"),
 

diff --git a/dinky-common/src/main/java/org/dinky/data/model/SystemConfiguration.java b/dinky-common/src/main/java/org/dinky/data/model/SystemConfiguration.java
@@ -265,6 +265,14 @@ public static Configuration.OptionBuilder key(Status status) {
             .stringType()
             .defaultValue("file:///")
             .note(Status.SYS_RESOURCE_SETTINGS_HDFS_FS_DEFAULTFS_NOTE);
+    private final Configuration<String> resourcesHdfsCoreSite = key(Status.SYS_RESOURCE_SETTINGS_HDFS_CORE_SITE)
+            .stringType()
+            .defaultValue("")
+            .note(Status.SYS_RESOURCE_SETTINGS_HDFS_CORE_SITE_NOTE);
+    private final Configuration<String> resourcesHdfsHdfsSite = key(Status.SYS_RESOURCE_SETTINGS_HDFS_HDFS_SITE)
+            .stringType()
+            .defaultValue("")
+            .note(Status.SYS_RESOURCE_SETTINGS_HDFS_HDFS_SITE_NOTE);
     private final Configuration<Boolean> resourcesPathStyleAccess = key(Status.SYS_RESOURCE_SETTINGS_PATH_STYLE_ACCESS)
             .booleanType()
             .defaultValue(true)

diff --git a/dinky-common/src/main/resources/i18n/messages_en_US.properties b/dinky-common/src/main/resources/i18n/messages_en_US.properties
@@ -260,6 +260,10 @@ sys.resource.settings.hdfs.root.user=HDFS operation user name
 sys.resource.settings.hdfs.root.user.note=HDFS operation user name
 sys.resource.settings.hdfs.fs.defaultFS=HDFS defaultFS
 sys.resource.settings.hdfs.fs.defaultFS.note=fs.defaultFS configuration items, such as remote: hdfs://localhost:9000, local: file:///
+sys.resource.settings.hdfs.core.site=core-site.xml
+sys.resource.settings.hdfs.core.site.note=core-site.xml configuration file content,High availability required
+sys.resource.settings.hdfs.hdfs.site=hdfs-site.xml
+sys.resource.settings.hdfs.hdfs.site.note=hdfs-site.xml configuration file content\uFF0CHigh availability required
 
 
 #Dinky Gateway
@@ -279,4 +283,4 @@ process.submit.execute.flinkSql=excute flinkSql
 process.register.exits=The current task is executing, do not submit it repeatedly, if you have any questions, please go to the configuration center to view
 # resource
 resource.root.dir.not.allow.delete=The root directory is not allowed to be deleted
-resource.dir.or.file.not.exist=The directory or file does not exist
+resource.dir.or.file.not.exist=The directory or file does not exist