Skip to content

AWS: Add check to create staging directory if not exists for S3OutputStream #3175

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

Merged
jackye1995 merged 6 commits into from
Sep 28, 2021
Merged

AWS: Add check to create staging directory if not exists for S3OutputStream #3175

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
30f52b1
AWS: Add check to create staging directory if not exists for S3Output…
rajarshisarkar Sep 24, 2021
05de05e
Addressing review comments
rajarshisarkar Sep 27, 2021
0ef2b28
Addressing review comments
rajarshisarkar Sep 27, 2021
523b701
Addressed review comments
rajarshisarkar Sep 28, 2021
cca5080
Addressed review comments
rajarshisarkar Sep 28, 2021
4cb6a0c
Addressed review comments
rajarshisarkar Sep 28, 2021
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
2 changes: 2 additions & 0 deletions api/src/main/java/org/apache/iceberg/io/OutputFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.apache.iceberg.io;

import java.io.IOException;
import java.security.AccessControlException;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.exceptions.RuntimeIOException;

Expand Down Expand Up @@ -48,6 +49,7 @@ public interface OutputFile {
*
* @return an output stream that can report its position
* @throws RuntimeIOException If the implementation throws an {@link IOException}
* @throws AccessControlException If the implementation throws an {@link SecurityException}

@jackye1995 jackye1995 Sep 28, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

as discussed in the block below, we can just document @throws SecurityException if staging directory creation fails due to missing JVM level permission.

@rajarshisarkar rajarshisarkar Sep 28, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

I have made the changes accordingly.

*/
PositionOutputStream createOrOverwrite();

Expand Down
4 changes: 4 additions & 0 deletions aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputFile.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@

import java.io.IOException;
import java.io.UncheckedIOException;
import java.security.AccessControlException;
import org.apache.iceberg.aws.AwsProperties;
import org.apache.iceberg.exceptions.AlreadyExistsException;
import org.apache.iceberg.io.InputFile;
Expand Down Expand Up @@ -62,6 +63,9 @@ public PositionOutputStream createOrOverwrite() {
return new S3OutputStream(client(), uri(), awsProperties());
} catch (IOException e) {
throw new UncheckedIOException("Failed to create output stream for location: " + uri(), e);
} catch (SecurityException e) {

@jackye1995 jackye1995 Sep 28, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

as discussed in the block below, we can remove this catch. AccessControlException also hides the original exception, so better to not use it.

@rajarshisarkar rajarshisarkar Sep 28, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Removed the special handling for SecurityException.

throw new AccessControlException(
"Access denied while creating staging directory for output stream: " + uri());
}
}

Expand Down
15 changes: 13 additions & 2 deletions aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -86,7 +86,8 @@ class S3OutputStream extends PositionOutputStream {
private boolean closed = false;

@SuppressWarnings("StaticAssignmentInConstructor")
S3OutputStream(S3Client s3, S3URI location, AwsProperties awsProperties) throws IOException {
S3OutputStream(S3Client s3, S3URI location, AwsProperties awsProperties)
throws IOException, SecurityException {

@jackye1995 jackye1995 Sep 28, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

same comment as below, remove SecurityException after throws

@rajarshisarkar rajarshisarkar Sep 28, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Removed it.

if (executorService == null) {
synchronized (S3OutputStream.class) {
if (executorService == null) {
Expand Down Expand Up @@ -170,11 +171,12 @@ public void write(byte[] b, int off, int len) throws IOException {
}
}

private void newStream() throws IOException {
private void newStream() throws IOException, SecurityException {

@jackye1995 jackye1995 Sep 28, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Sorry for the back and forth, I read the documentation for mkdirs, it seems like SecurityException only catches JVM level permission and it might still just return false for OS level permission failure, so it's hard to have a consistent behavior for error handling. Because of that, plus the fact that SecurityException is a runtime exception, I think we can remove the special handling of it and just let it throw to the top level.

@rajarshisarkar rajarshisarkar Sep 28, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Removed the special handling for SecurityException.

@jackye1995 jackye1995 Sep 28, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Thanks, can you also remove the SecurityException after throws? We don't need to throw runtime exception explicitly, just need to document it at top level which you already did.

@rajarshisarkar rajarshisarkar Sep 28, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Yeah, need not to throw explicitly. Removed it.

if (stream != null) {
stream.close();
}

createStagingDirectoryIfNotExists();
currentStagingFile = File.createTempFile("s3fileio-", ".tmp", stagingDirectory);
currentStagingFile.deleteOnExit();
stagingFiles.add(currentStagingFile);
Expand Down Expand Up @@ -328,6 +330,15 @@ private static InputStream uncheckedInputStream(File file) {
}
}

private void createStagingDirectoryIfNotExists() throws SecurityException {
if (!stagingDirectory.exists()) {

@jackye1995 jackye1995 Sep 28, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

add a LOG saying something like "staging directoy {} not exist, trying to create one"

@rajarshisarkar rajarshisarkar Sep 28, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Added the logger.

boolean createdStagingDirectory = stagingDirectory.mkdirs();
if (createdStagingDirectory) {
LOG.info("Successfully created staging directory: {}", stagingDirectory.getAbsolutePath());
}

@jackye1995 jackye1995 Sep 28, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

else, add a LOG saying "Staging directory {} creation failed, or it is created by another process"

@jackye1995 jackye1995 Sep 28, 2021

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

actually, we can do better than just logging. When creation fails, we can check directory existence again, if it exists then it's created by another process, otherwise it's still an issue, and we can throw an IOException with error message indicating staging directory creation fails for some unknown reasons.

@rajarshisarkar rajarshisarkar Sep 28, 2021

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

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

Agreed, I have made the changes accordingly.

}
}

@SuppressWarnings("checkstyle:NoFinalizer")
@Override
protected void finalize() throws Throwable {
Expand Down
19 changes: 19 additions & 0 deletions aws/src/test/java/org/apache/iceberg/aws/s3/S3OutputStreamTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.apache.iceberg.aws.s3;

import com.adobe.testing.s3mock.junit4.S3MockRule;
import java.io.File;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.file.Files;
Expand All @@ -29,6 +30,7 @@
import java.util.stream.Stream;
import org.apache.iceberg.aws.AwsProperties;
import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
import org.junit.After;
import org.junit.Before;
import org.junit.ClassRule;
import org.junit.Test;
Expand Down Expand Up @@ -72,6 +74,7 @@ public class S3OutputStreamTest {
private final S3Client s3mock = mock(S3Client.class, delegatesTo(s3));
private final Random random = new Random(1);
private final Path tmpDir = Files.createTempDirectory("s3fileio-test-");
private final String newTmpDirectory = "/tmp/newStagingDirectory";

private final AwsProperties properties = new AwsProperties(ImmutableMap.of(
AwsProperties.S3FILEIO_MULTIPART_SIZE, Integer.toString(5 * 1024 * 1024),
Expand All @@ -85,6 +88,14 @@ public void before() {
s3.createBucket(CreateBucketRequest.builder().bucket(BUCKET).build());
}

@After
public void after() {
File newStagingDirectory = new File(newTmpDirectory);
if (newStagingDirectory.exists()) {
newStagingDirectory.delete();
}
}

@Test
public void testWrite() {
// Run tests for both byte and array write paths
Expand Down Expand Up @@ -140,6 +151,14 @@ public void testMultipleClose() throws IOException {
stream.close();
}

@Test
public void testStagingDirectoryCreation() throws IOException {
AwsProperties newStagingDirectoryAwsProperties = new AwsProperties(ImmutableMap.of(
AwsProperties.S3FILEIO_STAGING_DIRECTORY, newTmpDirectory));
S3OutputStream stream = new S3OutputStream(s3, randomURI(), newStagingDirectoryAwsProperties);
stream.close();
}

private void writeAndVerify(S3Client client, S3URI uri, byte [] data, boolean arrayWrite) {
try (S3OutputStream stream = new S3OutputStream(client, uri, properties)) {
if (arrayWrite) {
Expand Down