From c0f3f4f0f996352737ffa40e428437d77f1a2cd2 Mon Sep 17 00:00:00 2001
From: Jongyoung Park <eminency@gmail.com>
Date: Fri, 9 Jan 2015 01:22:44 +0900
Subject: [PATCH] IOUtils.cleanup is used instead of direct close() in
 appenders.

---
 .../src/main/java/org/apache/tajo/storage/CSVFile.java        | 4 +---
 .../src/main/java/org/apache/tajo/storage/RowFile.java        | 3 ++-
 .../main/java/org/apache/tajo/storage/avro/AvroAppender.java  | 3 ++-
 .../java/org/apache/tajo/storage/parquet/ParquetAppender.java | 3 ++-
 .../tajo/storage/sequencefile/SequenceFileAppender.java       | 4 +---
 .../java/org/apache/tajo/storage/text/DelimitedTextFile.java  | 4 +---
 6 files changed, 9 insertions(+), 12 deletions(-)

diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/CSVFile.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/CSVFile.java
index 28c263c7b8..d4dde28000 100644
--- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/CSVFile.java
+++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/CSVFile.java
@@ -225,10 +225,8 @@ public void close() throws IOException {
           deflateFilter.resetState();
           deflateFilter = null;
         }
-
-        os.close();
       } finally {
-        IOUtils.cleanup(LOG, fos);
+        IOUtils.cleanup(LOG, os, fos, outputStream);
         if (compressor != null) {
           CodecPool.returnCompressor(compressor);
           compressor = null;
diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RowFile.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RowFile.java
index 5510cbf0cf..1ff6c4f9ea 100644
--- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RowFile.java
+++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/RowFile.java
@@ -25,6 +25,7 @@
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IOUtils;
 import org.apache.tajo.TaskAttemptId;
 import org.apache.tajo.catalog.Column;
 import org.apache.tajo.catalog.Schema;
@@ -468,7 +469,7 @@ public void close() throws IOException {
         }
         sync();
         out.flush();
-        out.close();
+        IOUtils.cleanup(LOG, out);
       }
     }
 
diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/avro/AvroAppender.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/avro/AvroAppender.java
index 69399dc0e0..0c34e9d32f 100644
--- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/avro/AvroAppender.java
+++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/avro/AvroAppender.java
@@ -28,6 +28,7 @@
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IOUtils;
 import org.apache.tajo.TaskAttemptId;
 import org.apache.tajo.catalog.Column;
 import org.apache.tajo.catalog.TableMeta;
@@ -201,7 +202,7 @@ public void flush() throws IOException {
    */
   @Override
   public void close() throws IOException {
-    dataFileWriter.close();
+    IOUtils.cleanup(null, dataFileWriter);
   }
 
   /**
diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/parquet/ParquetAppender.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/parquet/ParquetAppender.java
index ef5203c6d1..173c5bfa7e 100644
--- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/parquet/ParquetAppender.java
+++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/parquet/ParquetAppender.java
@@ -18,6 +18,7 @@
 
 package org.apache.tajo.storage.parquet;
 
+import org.apache.hadoop.io.IOUtils;
 import org.apache.tajo.TaskAttemptId;
 import org.apache.tajo.storage.StorageConstants;
 import parquet.hadoop.ParquetOutputFormat;
@@ -128,7 +129,7 @@ public void flush() throws IOException {
    */
   @Override
   public void close() throws IOException {
-    writer.close();
+    IOUtils.cleanup(null, writer);
   }
 
   public long getEstimatedOutputSize() throws IOException {
diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileAppender.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileAppender.java
index 8b5d677c44..404352c3e8 100644
--- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileAppender.java
+++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/sequencefile/SequenceFileAppender.java
@@ -248,7 +248,6 @@ public long getOffset() throws IOException {
   @Override
   public void flush() throws IOException {
     os.flush();
-    writer.close();
   }
 
   @Override
@@ -258,8 +257,7 @@ public void close() throws IOException {
       stats.setNumBytes(getOffset());
     }
 
-    os.close();
-    writer.close();
+    IOUtils.cleanup(LOG, os, writer);
   }
 
   @Override
diff --git a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java
index 15db4c335a..ac7c549cbd 100644
--- a/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java
+++ b/tajo-storage/tajo-storage-hdfs/src/main/java/org/apache/tajo/storage/text/DelimitedTextFile.java
@@ -241,10 +241,8 @@ public void close() throws IOException {
           deflateFilter.resetState();
           deflateFilter = null;
         }
-
-        os.close();
       } finally {
-        IOUtils.cleanup(LOG, fos);
+        IOUtils.cleanup(LOG, fos, os, outputStream);
         if (compressor != null) {
           CodecPool.returnCompressor(compressor);
           compressor = null;