diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/FSDataInputStreamWrapper.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/FSDataInputStreamWrapper.java index 67bca846521..c1f9a7ddacc 100644 --- a/hbase-server/src/main/java/org/apache/hadoop/hbase/io/FSDataInputStreamWrapper.java +++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/io/FSDataInputStreamWrapper.java @@ -20,11 +20,10 @@ package org.apache.hadoop.hbase.io; import java.io.Closeable; import java.io.IOException; import java.io.InputStream; -import java.lang.reflect.InvocationTargetException; -import java.lang.reflect.Method; import java.util.concurrent.atomic.AtomicInteger; import org.apache.commons.io.IOUtils; +import org.apache.hadoop.fs.CanUnbuffer; import org.apache.hadoop.fs.FSDataInputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; @@ -91,11 +90,6 @@ public class FSDataInputStreamWrapper implements Closeable { // reads without hbase checksum verification. private AtomicInteger hbaseChecksumOffCount = new AtomicInteger(-1); - private Boolean instanceOfCanUnbuffer = null; - // Using reflection to get org.apache.hadoop.fs.CanUnbuffer#unbuffer method to avoid compilation - // errors against Hadoop pre 2.6.4 and 2.7.1 versions. - private Method unbuffer = null; - private final static ReadStatistics readStatistics = new ReadStatistics(); private static class ReadStatistics { @@ -105,6 +99,9 @@ public class FSDataInputStreamWrapper implements Closeable { long totalZeroCopyBytesRead; } + private Boolean instanceOfCanUnbuffer = null; + private CanUnbuffer unbuffer = null; + public FSDataInputStreamWrapper(FileSystem fs, Path path) throws IOException { this(fs, path, false, -1L); } @@ -331,39 +328,23 @@ public class FSDataInputStreamWrapper implements Closeable { if (this.instanceOfCanUnbuffer == null) { // To ensure we compute whether the stream is instance of CanUnbuffer only once. this.instanceOfCanUnbuffer = false; - Class[] streamInterfaces = streamClass.getInterfaces(); - for (Class c : streamInterfaces) { - if (c.getCanonicalName().toString().equals("org.apache.hadoop.fs.CanUnbuffer")) { - try { - this.unbuffer = streamClass.getDeclaredMethod("unbuffer"); - } catch (NoSuchMethodException | SecurityException e) { - if (isLogTraceEnabled) { - LOG.trace("Failed to find 'unbuffer' method in class " + streamClass - + " . So there may be a TCP socket connection " - + "left open in CLOSE_WAIT state.", e); - } - return; - } - this.instanceOfCanUnbuffer = true; - break; - } + if (wrappedStream instanceof CanUnbuffer) { + this.unbuffer = (CanUnbuffer) wrappedStream; + this.instanceOfCanUnbuffer = true; } } if (this.instanceOfCanUnbuffer) { try { - this.unbuffer.invoke(wrappedStream); - } catch (IllegalAccessException | IllegalArgumentException | InvocationTargetException e) { + this.unbuffer.unbuffer(); + } catch (UnsupportedOperationException e){ if (isLogTraceEnabled) { LOG.trace("Failed to invoke 'unbuffer' method in class " + streamClass - + " . So there may be a TCP socket connection left open in CLOSE_WAIT state.", e); + + " . So there may be the stream does not support unbuffering.", e); } } } else { if (isLogTraceEnabled) { - LOG.trace("Failed to find 'unbuffer' method in class " + streamClass - + " . So there may be a TCP socket connection " - + "left open in CLOSE_WAIT state. For more details check " - + "https://issues.apache.org/jira/browse/HBASE-9393"); + LOG.trace("Failed to find 'unbuffer' method in class " + streamClass); } } } diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/io/TestFSDataInputStreamWrapper.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/TestFSDataInputStreamWrapper.java new file mode 100644 index 00000000000..22b6c62b486 --- /dev/null +++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/io/TestFSDataInputStreamWrapper.java @@ -0,0 +1,139 @@ +/** + * 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.hbase.io; + +import static org.junit.Assert.assertTrue; + +import java.io.IOException; +import java.io.InputStream; +import java.nio.ByteBuffer; +import java.util.EnumSet; + +import org.apache.hadoop.fs.ByteBufferReadable; +import org.apache.hadoop.fs.CanSetDropBehind; +import org.apache.hadoop.fs.CanSetReadahead; +import org.apache.hadoop.fs.CanUnbuffer; +import org.apache.hadoop.fs.FSDataInputStream; +import org.apache.hadoop.fs.FSInputStream; +import org.apache.hadoop.fs.HasEnhancedByteBufferAccess; +import org.apache.hadoop.fs.ReadOption; +import org.apache.hadoop.hbase.HBaseClassTestRule; +import org.apache.hadoop.hbase.testclassification.SmallTests; +import org.apache.hadoop.io.ByteBufferPool; +import org.junit.ClassRule; +import org.junit.Test; +import org.junit.experimental.categories.Category; + +@Category(SmallTests.class) +public class TestFSDataInputStreamWrapper { + + @ClassRule + public static final HBaseClassTestRule CLASS_RULE = + HBaseClassTestRule.forClass(TestFSDataInputStreamWrapper.class); + + @Test + public void testUnbuffer() throws Exception { + InputStream pc = new ParentClass(); + FSDataInputStreamWrapper fsdisw1 = + new FSDataInputStreamWrapper(new FSDataInputStream(pc)); + fsdisw1.unbuffer(); + // parent class should be true + assertTrue(((ParentClass)pc).getIsCallUnbuffer()); + fsdisw1.close(); + + InputStream cc1 = new ChildClass1(); + FSDataInputStreamWrapper fsdisw2 = + new FSDataInputStreamWrapper(new FSDataInputStream(cc1)); + fsdisw2.unbuffer(); + // child1 class should be true + assertTrue(((ChildClass1)cc1).getIsCallUnbuffer()); + fsdisw2.close(); + } + + private class ParentClass extends FSInputStream + implements ByteBufferReadable, CanSetDropBehind, CanSetReadahead, + HasEnhancedByteBufferAccess, CanUnbuffer { + + public boolean isCallUnbuffer = false; + + public boolean getIsCallUnbuffer(){ + return isCallUnbuffer; + } + + @Override + public void unbuffer() { + isCallUnbuffer = true; + } + + @Override + public int read() throws IOException { + return 0; + } + + @Override + public ByteBuffer read(ByteBufferPool paramByteBufferPool, + int paramInt, EnumSet paramEnumSet) + throws IOException, UnsupportedOperationException { + return null; + } + + @Override + public void releaseBuffer(ByteBuffer paramByteBuffer) { + + } + + @Override + public void setReadahead(Long paramLong) + throws IOException, UnsupportedOperationException { + + } + + @Override + public void setDropBehind(Boolean paramBoolean) + throws IOException, UnsupportedOperationException { + + } + + @Override + public int read(ByteBuffer paramByteBuffer) throws IOException { + return 0; + } + + @Override + public void seek(long paramLong) throws IOException { + + } + + @Override + public long getPos() throws IOException { + return 0; + } + + @Override + public boolean seekToNewSource(long paramLong) throws IOException { + return false; + } + } + + private class ChildClass1 extends ParentClass{ + @Override + public void unbuffer() { + isCallUnbuffer = true; + } + } +}