Adding AlreadyClosedException. Fixing formatting.

Author: kevins-29

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/AlreadyClosedException.java

@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.io.compress;
+
+import java.io.IOException;
+
+/**
+ * An exception class for when a closed compressor/decopressor is being used
+ * {@link org.apache.hadoop.io.compress.Compressor}
+ * {@link org.apache.hadoop.io.compress.Decompressor}
+ */
+public class AlreadyClosedException extends IOException {
+
+    /**
+     * Constructs a new exception with the specified cause and a detail
+     * message of <tt>(cause==null ? null : cause.toString())</tt> (which
+     * typically contains the class and detail message of <tt>cause</tt>).
+     * @param  cause the cause (which is saved for later retrieval by the
+     *         {@link #getCause()} method).  (A <tt>null</tt> value is
+     *         permitted, and indicates that the cause is nonexistent or
+     *         unknown.)
+     */
+    public AlreadyClosedException(Throwable cause) {
+        super(cause);
+    }
+}

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zlib/BuiltInGzipCompressor.java

@@ -24,6 +24,7 @@
 import java.util.zip.GZIPOutputStream;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.compress.AlreadyClosedException;
 import org.apache.hadoop.io.compress.Compressor;
 import org.apache.hadoop.io.compress.DoNotPool;
 import org.apache.hadoop.util.DataChecksum;
@@ -102,7 +103,15 @@ public int compress(byte[] b, int off, int len) throws IOException {
 
     if (state == BuiltInGzipDecompressor.GzipStateLabel.INFLATE_STREAM) {
       // now compress it into b[]
-      int deflated = deflater.deflate(b, off, len);
+      int deflated;
+      try {
+        deflated = deflater.deflate(b, off, len);
+      } catch (NullPointerException npe) {
+        if ("Deflater has been closed".equals(npe.getMessage())) {
+          throw new AlreadyClosedException(npe);
+        }
+        throw npe;
+      }
 
       compressedBytesWritten += deflated;
       off += deflated;

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/io/compress/zlib/BuiltInGzipDecompressor.java

@@ -23,6 +23,7 @@
 import java.util.zip.DataFormatException;
 import java.util.zip.Inflater;
 
+import org.apache.hadoop.io.compress.AlreadyClosedException;
 import org.apache.hadoop.io.compress.Decompressor;
 import org.apache.hadoop.io.compress.DoNotPool;
 import org.apache.hadoop.util.DataChecksum;
@@ -211,6 +212,11 @@ public synchronized int decompress(byte[] b, int off, int len)
         numAvailBytes = inflater.inflate(b, off, len);
       } catch (DataFormatException dfe) {
         throw new IOException(dfe.getMessage());
+      } catch (NullPointerException npe) {
+        if ("Inflater has been closed".equals(npe.getMessage())) {
+          throw new AlreadyClosedException(npe);
+        }
+        throw npe;
       }
       crc.update(b, off, numAvailBytes);  // CRC-32 is on _uncompressed_ data
       if (inflater.finished()) {

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/io/compress/TestCodecPool.java

@@ -207,12 +207,14 @@ public void testDoNotPoolCompressorNotUseableAfterReturn() throws Exception {
     final Compressor compressor = new BuiltInGzipCompressor(new Configuration());
     CodecPool.returnCompressor(compressor);
 
-    final CompressionOutputStream outputStream = gzipCodec.createOutputStream(new ByteArrayOutputStream(), compressor);
+    final CompressionOutputStream outputStream =
+            gzipCodec.createOutputStream(new ByteArrayOutputStream(), compressor);
     LambdaTestUtils.intercept(
-            NullPointerException.class,
+            AlreadyClosedException.class,
             "Deflater has been closed",
-            "Compressor from Codec with @DoNotPool should not be useable after returning to CodecPool",
-            () -> outputStream.write(1));
+            "Compressor from Codec with @DoNotPool should not be " +
+                    "useable after returning to CodecPool",
+        () -> outputStream.write(1));
   }
 
   @Test(timeout = 10000)
@@ -233,15 +235,17 @@ public void testDoNotPoolDecompressorNotUseableAfterReturn() throws Exception {
     final byte[] gzipBytes = baos.toByteArray();
     final ByteArrayInputStream bais = new ByteArrayInputStream(gzipBytes);
 
-    // BuiltInGzipDecompressor is an explicit example of a Decompressor with the @DoNotPool annotation
+    // BuiltInGzipDecompressor is an explicit example of a Decompressor
+    // with the @DoNotPool annotation
     final Decompressor decompressor = new BuiltInGzipDecompressor();
     CodecPool.returnDecompressor(decompressor);
 
     final CompressionInputStream inputStream = gzipCodec.createInputStream(bais, decompressor);
     LambdaTestUtils.intercept(
-            NullPointerException.class,
+            AlreadyClosedException.class,
             "Inflater has been closed",
-            "Decompressor from Codec with @DoNotPool should not be useable after returning to CodecPool",
-            () -> inputStream.read());
+            "Decompressor from Codec with @DoNotPool should not be " +
+                    "useable after returning to CodecPool",
+        () -> inputStream.read());
   }
 }