This is an automated email from the ASF dual-hosted git repository.
FrankChen021 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 5cad7d306c0 fix: allow HDFS killer to remove LZ4 segments (#20029)
5cad7d306c0 is described below
commit 5cad7d306c02af28d9e4bf678c5774b571ff51fc
Author: Frank Chen <[email protected]>
AuthorDate: Tue Aug 18 19:47:34 2026 +0800
fix: allow HDFS killer to remove LZ4 segments (#20029)
---
.../druid/storage/hdfs/HdfsDataSegmentKiller.java | 27 +++++++++-------
.../storage/hdfs/HdfsDataSegmentKillerTest.java | 36 ++++++++++++++++++++++
2 files changed, 52 insertions(+), 11 deletions(-)
diff --git
a/extensions-core/hdfs-storage/src/main/java/org/apache/druid/storage/hdfs/HdfsDataSegmentKiller.java
b/extensions-core/hdfs-storage/src/main/java/org/apache/druid/storage/hdfs/HdfsDataSegmentKiller.java
index 0f5822312b0..a8ea5291709 100644
---
a/extensions-core/hdfs-storage/src/main/java/org/apache/druid/storage/hdfs/HdfsDataSegmentKiller.java
+++
b/extensions-core/hdfs-storage/src/main/java/org/apache/druid/storage/hdfs/HdfsDataSegmentKiller.java
@@ -29,6 +29,7 @@ import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.segment.loading.DataSegmentKiller;
import org.apache.druid.segment.loading.SegmentLoadingException;
import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.utils.CompressionUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
@@ -71,8 +72,9 @@ public class HdfsDataSegmentKiller implements
DataSegmentKiller
log.info("Killing segment[%s] mapped to path[%s]", segment.getId(),
segmentPath);
try (final FileSystem fs = segmentPath.getFileSystem(config)) {
- String filename = segmentPath.getName();
- if (!filename.endsWith(".zip")) {
+ final String filename = segmentPath.getName();
+ final CompressionUtils.Format compressionFormat =
CompressionUtils.Format.fromFileName(filename);
+ if (compressionFormat != CompressionUtils.Format.ZIP &&
compressionFormat != CompressionUtils.Format.LZ4) {
throw new SegmentLoadingException("Unknown file type[%s]",
segmentPath);
} else {
@@ -81,17 +83,20 @@ public class HdfsDataSegmentKiller implements
DataSegmentKiller
return;
}
- // There are 3 supported path formats:
+ // There are 3 supported path formats for each segment compression
format:
// -
hdfs://nn1/hdfs_base_directory/data_source_name/interval/version/shardNum/index.zip
// -
hdfs://nn1/hdfs_base_directory/data_source_name/interval/version/shardNum_index.zip
// -
hdfs://nn1/hdfs_base_directory/data_source_name/interval/version/shardNum_UUID_index.zip
- String[] zipParts = filename.split("_");
+ // The same formats with an index.lz4 suffix are also supported.
+ final String[] segmentParts = filename.split("_");
Path descriptorPath = new Path(segmentPath.getParent(),
"descriptor.json");
- if (zipParts.length > 1) {
- Preconditions.checkState(zipParts.length <= 3 &&
- StringUtils.isNumeric(zipParts[0]) &&
- "index.zip".equals(zipParts[zipParts.length
- 1]),
+ if (segmentParts.length > 1) {
+ Preconditions.checkState(segmentParts.length <= 3 &&
+ StringUtils.isNumeric(segmentParts[0]) &&
+ ("index" +
compressionFormat.getSuffix()).equals(
+ segmentParts[segmentParts.length - 1]
+ ),
"Unexpected segmentPath format [%s]",
segmentPath
);
@@ -99,8 +104,8 @@ public class HdfsDataSegmentKiller implements
DataSegmentKiller
segmentPath.getParent(),
org.apache.druid.java.util.common.StringUtils.format(
"%s_%sdescriptor.json",
- zipParts[0],
- zipParts.length == 2 ? "" : zipParts[1] + "_"
+ segmentParts[0],
+ segmentParts.length == 2 ? "" : segmentParts[1] + "_"
)
);
}
@@ -113,7 +118,7 @@ public class HdfsDataSegmentKiller implements
DataSegmentKiller
// anymore, but we still delete them if exists.
fs.delete(descriptorPath, false);
- removeEmptyParentDirectories(fs, segmentPath, zipParts.length > 1 ? 2
: 3);
+ removeEmptyParentDirectories(fs, segmentPath, segmentParts.length > 1
? 2 : 3);
}
}
catch (IOException e) {
diff --git
a/extensions-core/hdfs-storage/src/test/java/org/apache/druid/storage/hdfs/HdfsDataSegmentKillerTest.java
b/extensions-core/hdfs-storage/src/test/java/org/apache/druid/storage/hdfs/HdfsDataSegmentKillerTest.java
index 2bbbf5d5b06..8cfb2239815 100644
---
a/extensions-core/hdfs-storage/src/test/java/org/apache/druid/storage/hdfs/HdfsDataSegmentKillerTest.java
+++
b/extensions-core/hdfs-storage/src/test/java/org/apache/druid/storage/hdfs/HdfsDataSegmentKillerTest.java
@@ -183,6 +183,42 @@ public class HdfsDataSegmentKillerTest
Assertions.assertTrue(fs.delete(dataSourceDir, false));
}
+ @Test
+ public void testKillLz4Segment() throws Exception
+ {
+ final File testRoot = FileUtils.createTempDir();
+ final Configuration config = new Configuration();
+ final HdfsDataSegmentKiller killer = new HdfsDataSegmentKiller(
+ config,
+ new HdfsDataSegmentPusherConfig()
+ {
+ @Override
+ public String getStorageDirectory()
+ {
+ return testRoot.getAbsolutePath();
+ }
+ }
+ );
+
+ final FileSystem fs = FileSystem.get(config);
+ final Path versionDir = new Path(testRoot.getAbsolutePath(),
"dataSource/interval/v1");
+ final Path segmentPath = new Path(versionDir, "3_index.lz4");
+ try {
+ Assertions.assertTrue(fs.mkdirs(versionDir));
+ fs.createNewFile(segmentPath);
+
+ killer.kill(getSegmentWithPath(segmentPath.toString()));
+
+ Assertions.assertFalse(fs.exists(segmentPath));
+ Assertions.assertFalse(fs.exists(versionDir));
+ Assertions.assertFalse(fs.exists(versionDir.getParent()));
+ Assertions.assertTrue(fs.exists(new Path(testRoot.getAbsolutePath(),
"dataSource")));
+ }
+ finally {
+ fs.delete(new Path(testRoot.getAbsolutePath()), true);
+ }
+ }
+
@Test
public void testKillNonExistingSegment() throws Exception
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
