[orc] branch main updated: ORC-978: Fix NPE in TestFlinkOrcReaderWriter (#891)

[do...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

dongjoon pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/orc.git


The following commit(s) were added to refs/heads/main by this push:
     new 8a4e24a  ORC-978: Fix NPE in TestFlinkOrcReaderWriter (#891)
8a4e24a is described below

commit 8a4e24a8144297c89eccb8a626cfe32a0af0e04e
Author: guiyanakaung <gu...@gmail.com>
AuthorDate: Thu Sep 2 08:18:04 2021 +0800

    ORC-978: Fix NPE in TestFlinkOrcReaderWriter (#891)
    
    ### What changes were proposed in this pull request?
    
    This PR aims to fix NPE in TestFlinkOrcReaderWriter.
    
    ```java
    @Override
      public void close() throws IOException {
        if (!isClosed) {
          try {
            if (batch.size > 0) {
              writer.addRowBatch(batch);
              batch.reset();
            }
          } finally {
            writer.close();
            this.isClosed = true;
          }
        }
      }
    ```
    
    writer.addRowBatch(batch);
    
    ```java
        .......
          checkMemory();
        } catch (Throwable t) {
          try {
            close();
          } catch (Throwable ignore) {
            // ignore
          }
          if (t instanceof IOException) {
            throw (IOException) t;
          } else {
            throw new IOException("Problem adding row to " + path, t);
          }
        }
    ```
    
    addRowBatch method throws java.lang.OutOfMemoryError causing writerImpl to close twice in TestFlinkOrcReaderWriter case.
    The first close already set the rawWriter to null, so the second time throw a NullPointerException.
    
    I added status variables to ensure that close will only do the necessary action the first time.
    
    ### Why are the changes needed?
    
    Fix NPE in TestFlinkOrcReaderWriter.
    
    ### How was this patch tested?
    
    Add UT for this bug.
---
 .../src/java/org/apache/orc/impl/WriterImpl.java   | 23 ++++++++++++++--------
 .../test/org/apache/orc/impl/TestWriterImpl.java   |  8 ++++++++
 2 files changed, 23 insertions(+), 8 deletions(-)

diff --git a/java/core/src/java/org/apache/orc/impl/WriterImpl.java b/java/core/src/java/org/apache/orc/impl/WriterImpl.java
index 969b3e2..d95430c 100644
--- a/java/core/src/java/org/apache/orc/impl/WriterImpl.java
+++ b/java/core/src/java/org/apache/orc/impl/WriterImpl.java
@@ -140,6 +140,7 @@ public class WriterImpl implements WriterInternal, MemoryManager.Callback {
   // information
   private boolean needKeyFlush;
   private final boolean useProlepticGregorian;
+  private boolean isClose = false;
 
   public WriterImpl(FileSystem fs,
                     Path path,
@@ -725,15 +726,21 @@ public class WriterImpl implements WriterInternal, MemoryManager.Callback {
 
   @Override
   public void close() throws IOException {
-    if (callback != null) {
-      callback.preFooterWrite(callbackContext);
+    if (!isClose) {
+      try {
+        if (callback != null) {
+          callback.preFooterWrite(callbackContext);
+        }
+        // remove us from the memory manager so that we don't get any callbacks
+        memoryManager.removeWriter(path);
+        // actually close the file
+        flushStripe();
+        lastFlushOffset = writeFooter();
+        physicalWriter.close();
+      } finally {
+        isClose = true;
+      }
     }
-    // remove us from the memory manager so that we don't get any callbacks
-    memoryManager.removeWriter(path);
-    // actually close the file
-    flushStripe();
-    lastFlushOffset = writeFooter();
-    physicalWriter.close();
   }
 
   /**
diff --git a/java/core/src/test/org/apache/orc/impl/TestWriterImpl.java b/java/core/src/test/org/apache/orc/impl/TestWriterImpl.java
index 712cd7e..b17e1b7 100644
--- a/java/core/src/test/org/apache/orc/impl/TestWriterImpl.java
+++ b/java/core/src/test/org/apache/orc/impl/TestWriterImpl.java
@@ -118,4 +118,12 @@ public class TestWriterImpl {
     Reader r = OrcFile.createReader(testFilePath, OrcFile.readerOptions(conf));
     assertEquals(r.getStripes(), w.getStripes());
   }
+
+  @Test
+  public void testCloseIsIdempotent() throws IOException {
+    conf.set(OrcConf.OVERWRITE_OUTPUT_FILE.getAttribute(), "true");
+    Writer w = OrcFile.createWriter(testFilePath, OrcFile.writerOptions(conf).setSchema(schema));
+    w.close();
+    w.close();
+  }
 }