GH-3350: Avoid flushing data to cloud when exception is thrown

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/InternalParquetRecordWriter.java

@@ -140,8 +140,9 @@ public void close() throws IOException, InterruptedException {
         }
         finalMetadata.putAll(finalWriteContext.getExtraMetaData());
         parquetFileWriter.end(finalMetadata);
+        AutoCloseables.uncheckedClose(parquetFileWriter);
       } finally {
-        AutoCloseables.uncheckedClose(columnStore, pageStore, bloomFilterWriteStore, parquetFileWriter);
+        AutoCloseables.uncheckedClose(columnStore, pageStore, bloomFilterWriteStore);
         closed = true;
       }
     }

parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java

@@ -113,7 +113,7 @@ public class ParquetFileWriter implements AutoCloseable {
   public static final byte[] EFMAGIC = EF_MAGIC_STR.getBytes(StandardCharsets.US_ASCII);
   public static final String PARQUET_COMMON_METADATA_FILE = "_common_metadata";
   public static final int CURRENT_VERSION = 1;
-
+  
   // File creation modes
   public static enum Mode {
     CREATE,
@@ -173,6 +173,7 @@ public static enum Mode {
 
   // set when end is called
   private ParquetMetadata footer = null;
+  private boolean aborted;
   private boolean closed;
 
   private final CRC32 crc;
@@ -1812,6 +1813,8 @@ public void end(Map<String, String> extraMetaData) throws IOException {
       LOG.debug("{}: end", out.getPos());
       this.footer = new ParquetMetadata(new FileMetaData(schema, extraMetaData, Version.FULL_VERSION), blocks);
       serializeFooter(footer, out, fileEncryptor, metadataConverter);
+    } catch (Exception e) {
+      aborted = true;
     } finally {
       close();
     }
@@ -1823,7 +1826,7 @@ public void close() throws IOException {
       return;
     }
     try (PositionOutputStream temp = out) {
-      temp.flush();
+      if (!aborted) temp.flush();
       if (crcAllocator != null) {
         crcAllocator.close();
       }

parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestParquetWriter.java

@@ -780,4 +780,44 @@ public void testParquetWriterBuilderCanNotConfigurePathAndFile() throws IOExcept
         "Cannot set both path and file", IllegalStateException.class, (Callable<ParquetWriter<Group>>) () ->
             ExampleParquetWriter.builder(path).withFile(outputFile).build());
   }
+
+  @Test
+  public void testNoFlushAfterException() throws Exception {
+    final File testDir = temp.newFile();
+    testDir.delete();
+
+    final Path file = new Path(testDir.getAbsolutePath(), "test.parquet");
+
+    MessageType schema = Types.buildMessage()
+        .required(BINARY)
+        .named("binary_field")
+        .required(INT32)
+        .named("int32_field")
+        .named("test_schema_abort");
+    Configuration conf = new Configuration();
+
+    try (ParquetWriter<Group> writer =  ExampleParquetWriter.builder(new Path(file.toString()))
+        .withAllocator(allocator)
+        .withType(schema)
+        .build()) {
+
+      SimpleGroupFactory f = new SimpleGroupFactory(schema);
+      writer.write(f.newGroup()
+          .append("binary_field", "hello")
+          .append("int32_field", 123));
+
+      Field internalWriterField = ParquetWriter.class.getDeclaredField("writer");
+      internalWriterField.setAccessible(true);
+      Object internalWriter = internalWriterField.get(writer);
+
+      Field abortedField = internalWriter.getClass().getDeclaredField("aborted");
+      abortedField.setAccessible(true);
+      abortedField.setBoolean(internalWriter, true);
+      writer.close();
+    }
+
+    // After closing, check whether file exists or is empty
+    FileSystem fs = file.getFileSystem(conf);
+    assertTrue(!fs.exists(file) || fs.getFileStatus(file).getLen() == 0);
+  }
 }