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

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

@@ -129,6 +129,7 @@ public void close() throws IOException, InterruptedException {
     if (!closed) {
       try {
         if (aborted) {
+          parquetFileWriter.abort();
           return;
         }
         flushRowGroupToStore();
@@ -140,6 +141,9 @@ public void close() throws IOException, InterruptedException {
         }
         finalMetadata.putAll(finalWriteContext.getExtraMetaData());
         parquetFileWriter.end(finalMetadata);
+      } catch (Exception e) {
+        parquetFileWriter.abort();
+        throw e;
       } finally {
         AutoCloseables.uncheckedClose(columnStore, pageStore, bloomFilterWriteStore, parquetFileWriter);
         closed = true;

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

@@ -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,22 +1813,35 @@ 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 (IOException e) {
+      abort();
+      throw e;
     } finally {
       close();
     }
   }
 
+  /* Mark the writer as aborted to avoid flushing incomplete data to the cloud. */
+  public void abort() {
+    aborted = true;
+  }
+
   @Override
   public void close() throws IOException {
     if (closed) {
       return;
     }
-    try (PositionOutputStream temp = out) {
-      temp.flush();
+
+    try {
+      if (!aborted && out != null) {
+        out.flush();
+      }
+    } catch (IOException e) {
+      throw e;
+    } finally {
       if (crcAllocator != null) {
         crcAllocator.close();
       }
-    } finally {
       closed = true;
     }
   }

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);
+  }
 }