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

rajarshisarkar · 2021-09-24T14:05:26Z

Fix for #2991

Seems like for EMR 6.x (Spark 3.x) Iceberg attempts to write the temp files (S3FileIO Reference) before the tmp directory is created in some cases. This results in java.io.IOException: No such file or directory exception.

The tmp files look something like this by default (here System.getProperty("java.io.tmpdir") is /mnt1/yarn/usercache/hadoop/appcache/application_1632368478936_0042/container_1632368478936_0042_01_000001/tmp):

/mnt1/yarn/usercache/hadoop/appcache/application_1632368478936_0042/container_1632368478936_0042_01_000001/tmp/s3fileio-2279038237918274355.tmp
/mnt1/yarn/usercache/hadoop/appcache/application_1632368478936_0042/container_1632368478936_0042_01_000001/tmp/s3fileio-2279038237918274355.tmp
/mnt1/yarn/usercache/hadoop/appcache/application_1632368478936_0042/container_1632368478936_0042_01_000001/tmp/s3fileio-2279038237918274355.tmp

(or) something like this if s3fileIoStagingDirectory is set via s3.staging-dir (/mnt/tmp for this case):

/mnt/tmp/s3fileio-6715538145691725671.tmp
/mnt/tmp/s3fileio-5266666342669697861.tmp
/mnt/tmp/s3fileio-4592572815837116260.tmp

Adding a defensive check to ensure we check the existence of the staging directory before writing the temp files. With this check I could see the executor tasks succeeding in cluster deploy mode consistently.

Configurations for testing:

EMR 6.4.0, Iceberg 0.12.0, Master node: 1 instance, Core node: 10 instances, Spark 3.1.2
EMR 6.2.0, Iceberg 0.11.1, Master node: 1 instance, Core node: 10 instances, Spark 3.0.1

Final Status of jobs deployed in cluster mode:

…Stream

rajarshisarkar · 2021-09-24T14:19:53Z

@thaismonti1912 @alex-shchetkov @fcvr1010 @jackye1995 Can you please review the PR.

jackye1995 · 2021-09-24T16:43:18Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java

@@ -175,8 +175,15 @@ private void newStream() throws IOException {
      stream.close();
    }

+    boolean createStagingDirectory = false;
+    if (!stagingDirectory.exists()) {
+      createStagingDirectory = stagingDirectory.mkdirs();


based on the documentation, this can also throw SecurityException that needs to be handled. We should also add some logging if staging directory is created, and if creation failed for some reason.

Handled SecurityException and added loggers for success scenario. Throwing the exception in case of failure scenario.

jackye1995 · 2021-09-24T16:43:38Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java

+    boolean createStagingDirectory = false;
+    if (!stagingDirectory.exists()) {
+      createStagingDirectory = stagingDirectory.mkdirs();
+    }


nit: newline needed after a control statement

Addressed. Also, separated out the logic in a method: createStagingDirectoryIfNotExists()

jackye1995 · 2021-09-24T16:43:42Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java

    currentStagingFile = File.createTempFile("s3fileio-", ".tmp", stagingDirectory);
    currentStagingFile.deleteOnExit();
+    if (createStagingDirectory) {
+      stagingDirectory.deleteOnExit();
+    }


nit: newline needed after a control statement

Addressed. Also, separated out the logic in a method: createStagingDirectoryIfNotExists()

jackye1995 · 2021-09-24T16:44:27Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java

    currentStagingFile = File.createTempFile("s3fileio-", ".tmp", stagingDirectory);
    currentStagingFile.deleteOnExit();
+    if (createStagingDirectory) {
+      stagingDirectory.deleteOnExit();


why do we need to delete on exit? I think we can assume it's safe to keep the staging directory there.

Yeah, it should be safe. Made the changes accordingly.

jackye1995 · 2021-09-24T16:45:02Z

Thanks for working on this! Can you also add a test case for it?

rajarshisarkar · 2021-09-27T11:14:57Z

Thanks for working on this! Can you also add a test case for it?

Added the test case testStagingDirectoryCreation() in S3OutputStreamTest

jackye1995 · 2021-09-28T06:50:56Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java

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


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

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.

Agreed, I have made the changes accordingly.

jackye1995 · 2021-09-28T06:50:58Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java

@@ -328,6 +330,15 @@ private static InputStream uncheckedInputStream(File file) {
    }
  }

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


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

Added the logger.

jackye1995 · 2021-09-28T06:53:12Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java

@@ -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 {


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.

Removed the special handling for SecurityException.

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.

Yeah, need not to throw explicitly. Removed it.

jackye1995 · 2021-09-28T06:53:58Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputFile.java

@@ -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) {


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

Removed the special handling for SecurityException.

jackye1995 · 2021-09-28T06:55:18Z

api/src/main/java/org/apache/iceberg/io/OutputFile.java

@@ -48,6 +49,7 @@
   *
   * @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}


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

I have made the changes accordingly.

jackye1995 · 2021-09-28T16:23:10Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java

+        LOG.info("Successfully created staging directory: {}", stagingDirectory.getAbsolutePath());
+      } else {
+        if (stagingDirectory.exists()) {
+          LOG.info("Staging directory: {} is created by another process",


nit: match the log message at L339, "Successfully created staging directory by another process: {}"

Made the change.

jackye1995 · 2021-09-28T16:24:20Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java

+              stagingDirectory.getAbsolutePath());
+        } else {
+          throw new IOException(String
+              .format("Staging directory: %s creation failed due to some unknown reason",


nit: match the log message at L346, "Fail to create staging directory due to some unknown reason: {}"

Made the change.

jackye1995 · 2021-09-28T16:30:45Z

aws/src/main/java/org/apache/iceberg/aws/s3/S3OutputStream.java

@@ -86,7 +86,8 @@
  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 {


same comment as below, remove SecurityException after throws

Removed it.

jackye1995 · 2021-09-28T16:31:21Z

Thank you for the work, looks good to me overall, just some styling changes then it should be good to go!

…Stream (apache#3175)

…Stream (#3175)

…Stream (apache#3175)

AWS: Add check to create staging directory if not exists for S3Output…

30f52b1

…Stream

github-actions bot added the AWS label Sep 24, 2021

rajarshisarkar mentioned this pull request Sep 24, 2021

No such file or directory when using multiple Spark executors + Iceberg in EMR #2991

Closed

jackye1995 reviewed Sep 24, 2021

View reviewed changes

Addressing review comments

05de05e

github-actions bot added the API label Sep 27, 2021

rajarshisarkar requested a review from jackye1995 September 27, 2021 11:26

Addressing review comments

0ef2b28

jackye1995 reviewed Sep 28, 2021

View reviewed changes

Addressed review comments

523b701

rajarshisarkar requested a review from jackye1995 September 28, 2021 13:44

jackye1995 reviewed Sep 28, 2021

View reviewed changes

Addressed review comments

cca5080

rajarshisarkar requested a review from jackye1995 September 28, 2021 17:30

Addressed review comments

4cb6a0c

jackye1995 approved these changes Sep 28, 2021

View reviewed changes

jackye1995 merged commit 1fde388 into apache:master Sep 28, 2021

kbendick added this to the Java 0.12.1 Release milestone Oct 26, 2021

kbendick pushed a commit to kbendick/iceberg that referenced this pull request Oct 27, 2021

AWS: Add check to create staging directory if not exists for S3Output…

ab88000

…Stream (apache#3175)

kbendick pushed a commit to kbendick/iceberg that referenced this pull request Oct 27, 2021

AWS: Add check to create staging directory if not exists for S3Output…

d5922f7

…Stream (apache#3175)

rdblue pushed a commit that referenced this pull request Oct 29, 2021

AWS: Add check to create staging directory if not exists for S3Output…

c739204

…Stream (#3175)

izchen pushed a commit to izchen/iceberg that referenced this pull request Dec 7, 2021

AWS: Add check to create staging directory if not exists for S3Output…

4869763

…Stream (apache#3175)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

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

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

rajarshisarkar commented Sep 24, 2021 •

edited

Loading

rajarshisarkar commented Sep 24, 2021

jackye1995 Sep 24, 2021

rajarshisarkar Sep 27, 2021 •

edited

Loading

jackye1995 Sep 24, 2021

rajarshisarkar Sep 27, 2021

jackye1995 Sep 24, 2021

rajarshisarkar Sep 27, 2021

jackye1995 Sep 24, 2021

rajarshisarkar Sep 27, 2021

jackye1995 commented Sep 24, 2021

rajarshisarkar commented Sep 27, 2021

jackye1995 Sep 28, 2021

jackye1995 Sep 28, 2021

rajarshisarkar Sep 28, 2021

jackye1995 Sep 28, 2021

rajarshisarkar Sep 28, 2021

jackye1995 Sep 28, 2021

rajarshisarkar Sep 28, 2021

jackye1995 Sep 28, 2021

rajarshisarkar Sep 28, 2021

jackye1995 Sep 28, 2021

rajarshisarkar Sep 28, 2021

jackye1995 Sep 28, 2021

rajarshisarkar Sep 28, 2021

jackye1995 Sep 28, 2021

rajarshisarkar Sep 28, 2021

jackye1995 Sep 28, 2021

rajarshisarkar Sep 28, 2021

jackye1995 Sep 28, 2021

rajarshisarkar Sep 28, 2021

jackye1995 commented Sep 28, 2021

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

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

Conversation

rajarshisarkar commented Sep 24, 2021 • edited Loading

rajarshisarkar commented Sep 24, 2021

Choose a reason for hiding this comment

rajarshisarkar Sep 27, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jackye1995 commented Sep 24, 2021

rajarshisarkar commented Sep 27, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jackye1995 commented Sep 28, 2021

rajarshisarkar commented Sep 24, 2021 •

edited

Loading

rajarshisarkar Sep 27, 2021 •

edited

Loading