Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

HdfsDataSegmentPusher: Properly include scheme, host in output path if necessary. #3577

Merged
merged 1 commit into from
Oct 17, 2016
Merged

HdfsDataSegmentPusher: Properly include scheme, host in output path if necessary. #3577

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 4 additions & 3 deletions extensions-core/hdfs-storage/src/main/java/io/druid/storage/hdfs/HdfsDataSegmentPusher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -103,9 +103,10 @@ public DataSegment push(File inDir, DataSegment segment) throws IOException
final DataSegment dataSegment;
try (FSDataOutputStream out = fs.create(tmpFile)) {
size = CompressionUtils.zip(inDir, out);
Path outDir = new Path(String.format("%s/%s", config.getStorageDirectory(), storageDir));
final Path outFile = new Path(String.format("%s/%s/index.zip", config.getStorageDirectory(), storageDir));
final Path outDir = outFile.getParent();
dataSegment = createDescriptorFile(
segment.withLoadSpec(makeLoadSpec(new Path(String.format("%s/%s", outDir.toUri().getPath(), "index.zip"))))
segment.withLoadSpec(makeLoadSpec(outFile))
.withSize(size)
.withBinaryVersion(SegmentUtils.getVersionFromDir(inDir)),
tmpFile.getParent(),
Expand Down Expand Up @@ -150,7 +151,7 @@ private DataSegment createDescriptorFile(DataSegment segment, Path outDir, final

private ImmutableMap<String, Object> makeLoadSpec(Path outFile)
{
return ImmutableMap.<String, Object>of("type", "hdfs", "path", outFile.toString());
return ImmutableMap.<String, Object>of("type", "hdfs", "path", outFile.toUri().toString());
Copy link
Member

@pjain1 pjain1 Oct 16, 2016 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

why was this not needed before ? It wasoutFile.toString() before #3494 and used to work.
FYI - I see this note in the toString() method of Path class -

// we can't use uri.toString(), which escapes everything, because we want
// illegal characters unescaped in the string, for glob processing, etc.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It helps the new unit test to pass; without it, file:///x/y/z gets translated to file:/x/y/z.

I think that comment on Path actually indicates that adding toUri is a good thing. We're adding a path to a single file and don't want glob processing or any other special processing.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok

}

private static class HdfsOutputStreamSupplier extends ByteSink
Expand Down
51 changes: 42 additions & 9 deletions ...ions-core/hdfs-storage/src/test/java/io/druid/storage/hdfs/HdfsDataSegmentPusherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -28,10 +28,12 @@
import io.druid.timeline.DataSegment;
import io.druid.timeline.partition.NoneShardSpec;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.joda.time.Interval;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.junit.rules.TemporaryFolder;

import java.io.File;
Expand All @@ -44,8 +46,33 @@ public class HdfsDataSegmentPusherTest
@Rule
public final TemporaryFolder tempFolder = new TemporaryFolder();

@Rule
public final ExpectedException expectedException = ExpectedException.none();

@Test
public void testPushWithScheme() throws Exception
{
testUsingScheme("file");
}

@Test
public void testPush() throws Exception
public void testPushWithBadScheme() throws Exception
{
expectedException.expect(IOException.class);
expectedException.expectMessage("No FileSystem for scheme: xyzzy");
testUsingScheme("xyzzy");

// Not reached
Assert.assertTrue(false);
}

@Test
public void testPushWithoutScheme() throws Exception
{
testUsingScheme(null);
}

private void testUsingScheme(final String scheme) throws Exception
{
Configuration conf = new Configuration(true);

Expand All @@ -58,8 +85,13 @@ public void testPush() throws Exception
final long size = data.length;

HdfsDataSegmentPusherConfig config = new HdfsDataSegmentPusherConfig();
final File storageDirectory = tempFolder.newFolder();

config.setStorageDirectory(tempFolder.newFolder().getAbsolutePath());
config.setStorageDirectory(
scheme != null
? String.format("%s://%s", scheme, storageDirectory.getAbsolutePath())
: storageDirectory.getAbsolutePath()
);
HdfsDataSegmentPusher pusher = new HdfsDataSegmentPusher(config, conf, new DefaultObjectMapper());

DataSegment segmentToPush = new DataSegment(
Expand All @@ -82,20 +114,21 @@ public void testPush() throws Exception
"type",
"hdfs",
"path",
String.format("%s/%s/index.zip",
config.getStorageDirectory(),
DataSegmentPusherUtil.getHdfsStorageDir(segmentToPush)
String.format(
"%s/%s/index.zip",
config.getStorageDirectory(),
DataSegmentPusherUtil.getHdfsStorageDir(segmentToPush)
)
), segment.getLoadSpec());
// rename directory after push
final String storageDir = DataSegmentPusherUtil.getHdfsStorageDir(segment);
File indexFile = new File(String.format("%s/%s/index.zip", config.getStorageDirectory(), storageDir));
final String segmentPath = DataSegmentPusherUtil.getHdfsStorageDir(segment);
File indexFile = new File(String.format("%s/%s/index.zip", storageDirectory, segmentPath));
Assert.assertTrue(indexFile.exists());
File descriptorFile = new File(String.format("%s/%s/descriptor.json", config.getStorageDirectory(), storageDir));
File descriptorFile = new File(String.format("%s/%s/descriptor.json", storageDirectory, segmentPath));
Assert.assertTrue(descriptorFile.exists());

// push twice will fail and temp dir cleaned
File outDir = new File(String.format("%s/%s", config.getStorageDirectory(), storageDir));
File outDir = new File(String.format("%s/%s", config.getStorageDirectory(), segmentPath));
outDir.setReadOnly();
try {
pusher.push(segmentDir, segmentToPush);
Expand Down