[GitHub] [incubator-druid] gianm commented on a change in pull request #7537: Avoid file rename when pushing segments with HDFSDataPusher

[GitBox <gi...@apache.org>]

gianm commented on a change in pull request #7537:  Avoid file rename when pushing segments with HDFSDataPusher
URL: https://github.com/apache/incubator-druid/pull/7537#discussion_r349886868
 
 

 ##########
 File path: extensions-core/hdfs-storage/src/main/java/org/apache/druid/storage/hdfs/HdfsDataSegmentPusher.java
 ##########
 @@ -107,70 +104,28 @@ public DataSegment push(final File inDir, final DataSegment segment, final boole
         fullyQualifiedStorageDirectory.get(),
         storageDir
     );
+    final String uniquePrefix = useUniquePath ? DataSegmentPusher.generateUniquePath() + "_" : "";
 
-    Path tmpIndexFile = new Path(StringUtils.format(
-        "%s/%s/%s/%s_index.zip",
+    final Path outIndexFile = new Path(StringUtils.format(
+        "%s/%s/%d_%sindex.zip",
         fullyQualifiedStorageDirectory.get(),
-        segment.getDataSource(),
-        UUIDUtils.generateUuid(),
-        segment.getShardSpec().getPartitionNum()
+        storageDir,
+        segment.getShardSpec().getPartitionNum(),
+        uniquePrefix
     ));
-    FileSystem fs = tmpIndexFile.getFileSystem(hadoopConfig);
-
-    fs.mkdirs(tmpIndexFile.getParent());
-    log.info("Compressing files from[%s] to [%s]", inDir, tmpIndexFile);
+    FileSystem fs = outIndexFile.getFileSystem(hadoopConfig);
+    // Create parent if it does not exist, recreation is not an error
+    fs.mkdirs(outIndexFile.getParent());
+    log.info("Compressing files from[%s] to [%s]", inDir, outIndexFile);
 
     final long size;
-    final DataSegment dataSegment;
-    try {
-      try (FSDataOutputStream out = fs.create(tmpIndexFile)) {
-        size = CompressionUtils.zip(inDir, out);
-      }
-
-      final String uniquePrefix = useUniquePath ? DataSegmentPusher.generateUniquePath() + "_" : "";
-      final Path outIndexFile = new Path(StringUtils.format(
-          "%s/%s/%d_%sindex.zip",
-          fullyQualifiedStorageDirectory.get(),
-          storageDir,
-          segment.getShardSpec().getPartitionNum(),
-          uniquePrefix
-      ));
-
-      dataSegment = segment.withLoadSpec(makeLoadSpec(outIndexFile.toUri()))
-                           .withSize(size)
-                           .withBinaryVersion(SegmentUtils.getVersionFromDir(inDir));
-
-      // Create parent if it does not exist, recreation is not an error
-      fs.mkdirs(outIndexFile.getParent());
-      copyFilesWithChecks(fs, tmpIndexFile, outIndexFile);
-    }
-    finally {
-      try {
-        if (fs.exists(tmpIndexFile.getParent()) && !fs.delete(tmpIndexFile.getParent(), true)) {
-          log.error("Failed to delete temp directory[%s]", tmpIndexFile.getParent());
-        }
-      }
-      catch (IOException ex) {
-        log.error(ex, "Failed to delete temp directory[%s]", tmpIndexFile.getParent());
-      }
+    try (FSDataOutputStream out = fs.create(outIndexFile, false)) {
 
 Review comment:
   I am very sad to say that I think this won't work for `index_realtime` tasks (used by Tranquility). Even though as a community we seem to be treating it as sunsetted, I think it'd still be good to not break it. Anyway, the problem is that RealtimePlumber pushes with `useUniquePath = false`, but can have replicas, and then this call will fail on one of the replicas.
   
   Thinking about the best way to address this, some options:
   
   1. Set `useUniquePath = true` in RealtimePlumber. Unfortunately, unlike the Appenderator system, it doesn't have logic to detect when it lost a publish race, meaning it would be tough to clean up the "extra" segments after a push. So this means wasted space on deep storage when replicas are enabled.
   2. Set overwrite to `true` in this `fs.create` call. This would be bad for any _other_ kind of ingestion (where we don't intend to overwrite and want fail-fast behavior if we do by accident). So I think probably a no go.
   3. Detect when a file already exists (based on the specific exception that gets thrown by `fs.create`), then throw a special exception callers can use to realize their segment already existed, or something like that. RealtimePlumber should catch that special exception and silently continue. Everything else should blow up on it (since it's not expected in any other case).
   
   I think (3) sounds best to me right now…

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


With regards,
Apache Git Services

---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org