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:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
