[hotfix-#1745][connector][hdfs][hive] Data loss occurs when multiple …

…tasks write to the same hdfs (or hive) data source
DTStack · Oct 30, 2023 · 69824c8 · 69824c8
1 parent e5c0066
commit 69824c8
Show file tree

Hide file tree

Showing 7 changed files with 47 additions and 3 deletions.
diff --git a/...ctor-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/sink/BaseHdfsOutputFormat.java b/...ctor-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/sink/BaseHdfsOutputFormat.java
@@ -131,7 +131,21 @@ protected void deleteDataDir() {
 
     @Override
     protected void deleteTmpDataDir() {
-        deleteDirectory(tmpPath);
+        try {
+            Path dir = new Path(outputFilePath);
+            if (fs == null) {
+                openSource();
+            }
+            FileStatus[] fileStatuses =
+                    fs.listStatus(dir, path -> path.getName().startsWith(TMP_DIR_NAME));
+            for (FileStatus fileStatus : fileStatuses) {
+                if (fileStatus.isDirectory()) {
+                    deleteDirectory(fileStatus.getPath().toString());
+                }
+            }
+        } catch (IOException e) {
+            throw new ChunJunRuntimeException("cannot delete tmpDir: ", e);
+        }
     }
 
     @Override

diff --git a/...-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/table/HdfsDynamicTableFactory.java b/...-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/table/HdfsDynamicTableFactory.java
@@ -135,6 +135,7 @@ private HdfsConfig getHdfsConfig(ReadableConfig config) {
         hdfsConfig.setEncoding(config.get(BaseFileOptions.ENCODING));
         hdfsConfig.setMaxFileSize(config.get(BaseFileOptions.MAX_FILE_SIZE));
         hdfsConfig.setNextCheckRows(config.get(BaseFileOptions.NEXT_CHECK_ROWS));
+        hdfsConfig.setJobTimeStamp(config.get(BaseFileOptions.JOB_TIMESTAMP));
 
         hdfsConfig.setDefaultFS(config.get(HdfsOptions.DEFAULT_FS));
         hdfsConfig.setFileType(config.get(HdfsOptions.FILE_TYPE));

diff --git a/...-hive/src/main/java/com/dtstack/chunjun/connector/hive/table/HiveDynamicTableFactory.java b/...-hive/src/main/java/com/dtstack/chunjun/connector/hive/table/HiveDynamicTableFactory.java
@@ -228,6 +228,7 @@ private HiveConfig getHiveConf(ReadableConfig config) {
         hiveConf.setEncoding(config.get(BaseFileOptions.ENCODING));
         hiveConf.setMaxFileSize(config.get(BaseFileOptions.MAX_FILE_SIZE));
         hiveConf.setNextCheckRows(config.get(BaseFileOptions.NEXT_CHECK_ROWS));
+        hiveConf.setJobTimeStamp(config.get(BaseFileOptions.JOB_TIMESTAMP));
 
         hiveConf.setDefaultFS(config.get(HdfsOptions.DEFAULT_FS));
         hiveConf.setFileType(config.get(HdfsOptions.FILE_TYPE));

diff --git a/...or-hive3/src/main/java/com/dtstack/chunjun/connector/hive3/sink/BaseHdfsOutputFormat.java b/...or-hive3/src/main/java/com/dtstack/chunjun/connector/hive3/sink/BaseHdfsOutputFormat.java
@@ -107,7 +107,21 @@ protected void deleteDataDir() {
 
     @Override
     protected void deleteTmpDataDir() {
-        deleteDirectory(tmpPath);
+        try {
+            Path dir = new Path(outputFilePath);
+            if (fs == null) {
+                openSource();
+            }
+            FileStatus[] fileStatuses =
+                    fs.listStatus(dir, path -> path.getName().startsWith(TMP_DIR_NAME));
+            for (FileStatus fileStatus : fileStatuses) {
+                if (fileStatus.isDirectory()) {
+                    deleteDirectory(fileStatus.getPath().toString());
+                }
+            }
+        } catch (IOException e) {
+            throw new ChunJunRuntimeException("cannot delete tmpDir: ", e);
+        }
     }
 
     @Override

diff --git a/chunjun-core/src/main/java/com/dtstack/chunjun/config/BaseFileConfig.java b/chunjun-core/src/main/java/com/dtstack/chunjun/config/BaseFileConfig.java
@@ -50,4 +50,6 @@ public class BaseFileConfig extends CommonConfig {
     private long nextCheckRows = 5000;
 
     private String suffix;
+
+    private String jobTimeStamp = "";
 }
diff --git a/chunjun-core/src/main/java/com/dtstack/chunjun/sink/format/BaseFileOutputFormat.java b/chunjun-core/src/main/java/com/dtstack/chunjun/sink/format/BaseFileOutputFormat.java
@@ -107,7 +107,12 @@ protected void initVariableFields() {
         } else {
             outputFilePath = baseFileConfig.getPath();
         }
-        tmpPath = outputFilePath + File.separatorChar + TMP_DIR_NAME;
+        tmpPath =
+                outputFilePath
+                        + File.separatorChar
+                        + TMP_DIR_NAME
+                        + baseFileConfig.getJobTimeStamp();
+        log.info("initVariableFields get tmpPath: {}", tmpPath);
         nextNumForCheckDataSize = baseFileConfig.getNextCheckRows();
         openSource();
     }

diff --git a/chunjun-core/src/main/java/com/dtstack/chunjun/table/options/BaseFileOptions.java b/chunjun-core/src/main/java/com/dtstack/chunjun/table/options/BaseFileOptions.java
@@ -62,4 +62,11 @@ public class BaseFileOptions {
                     .longType()
                     .defaultValue(5000L)
                     .withDescription("The number of data written in the next file size check");
+
+    public static final ConfigOption<String> JOB_TIMESTAMP =
+            ConfigOptions.key("properties.job-timestamp")
+                    .stringType()
+                    .defaultValue(String.valueOf(System.currentTimeMillis()))
+                    .withDescription(
+                            "To solve the problem of file loss caused by multiple tasks writing to the same output source at the same time");
 }