You are viewing a plain text version of this content. The canonical link for it is here.
Posted to common-commits@hadoop.apache.org by vi...@apache.org on 2017/05/11 21:42:10 UTC
[14/50] [abbrv] hadoop git commit: HDFS-11644. Support for querying
outputstream capabilities. Contributed by Manoj Govindassamy.
HDFS-11644. Support for querying outputstream capabilities. Contributed by Manoj Govindassamy.
Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/54fd0e44
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/54fd0e44
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/54fd0e44
Branch: refs/heads/HDFS-9806
Commit: 54fd0e44b76c4b982dcfb47932b6159851f14136
Parents: 749e5c0
Author: Andrew Wang <wa...@apache.org>
Authored: Mon May 8 21:59:49 2017 -0700
Committer: Andrew Wang <wa...@apache.org>
Committed: Mon May 8 21:59:49 2017 -0700
----------------------------------------------------------------------
.../apache/hadoop/fs/FSDataOutputStream.java | 10 ++-
.../apache/hadoop/fs/StreamCapabilities.java | 67 ++++++++++++++++++++
.../org/apache/hadoop/hdfs/DFSOutputStream.java | 15 ++++-
.../hadoop/hdfs/DFSStripedOutputStream.java | 11 +++-
.../apache/hadoop/hdfs/TestDFSOutputStream.java | 25 ++++++++
.../hadoop/hdfs/TestDFSStripedOutputStream.java | 24 ++++---
.../fs/azure/SyncableDataOutputStream.java | 12 +++-
7 files changed, 150 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/hadoop/blob/54fd0e44/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataOutputStream.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataOutputStream.java
index 3f97ea8..1d95cd3 100644
--- a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataOutputStream.java
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataOutputStream.java
@@ -30,7 +30,7 @@ import org.apache.hadoop.classification.InterfaceStability;
@InterfaceAudience.Public
@InterfaceStability.Stable
public class FSDataOutputStream extends DataOutputStream
- implements Syncable, CanSetDropBehind {
+ implements Syncable, CanSetDropBehind, StreamCapabilities {
private final OutputStream wrappedStream;
private static class PositionCache extends FilterOutputStream {
@@ -111,6 +111,14 @@ public class FSDataOutputStream extends DataOutputStream
return wrappedStream;
}
+ @Override
+ public boolean hasCapability(String capability) {
+ if (wrappedStream instanceof StreamCapabilities) {
+ return ((StreamCapabilities) wrappedStream).hasCapability(capability);
+ }
+ return false;
+ }
+
@Override // Syncable
public void hflush() throws IOException {
if (wrappedStream instanceof Syncable) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/54fd0e44/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java
----------------------------------------------------------------------
diff --git a/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java
new file mode 100644
index 0000000..65aa679
--- /dev/null
+++ b/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/StreamCapabilities.java
@@ -0,0 +1,67 @@
+/**
+ * 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.fs;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Interface to query streams for supported capabilities.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public interface StreamCapabilities {
+ /**
+ * Capabilities that a stream can support and be queried for.
+ */
+ enum StreamCapability {
+ /**
+ * Stream hflush capability to flush out the data in client's buffer.
+ * Streams with this capability implement {@link Syncable} and support
+ * {@link Syncable#hflush()}.
+ */
+ HFLUSH("hflush"),
+
+ /**
+ * Stream hsync capability to flush out the data in client's buffer and
+ * the disk device. Streams with this capability implement {@link Syncable}
+ * and support {@link Syncable#hsync()}.
+ */
+ HSYNC("hsync");
+
+ private final String capability;
+
+ StreamCapability(String value) {
+ this.capability = value;
+ }
+
+ public final String getValue() {
+ return capability;
+ }
+ }
+
+ /**
+ * Query the stream for a specific capability.
+ *
+ * @param capability string to query the stream support for.
+ * @return True if the stream supports capability.
+ */
+ boolean hasCapability(String capability);
+}
+
http://git-wip-us.apache.org/repos/asf/hadoop/blob/54fd0e44/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java
index ceaefd8..83f1425 100755
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java
@@ -17,6 +17,9 @@
*/
package org.apache.hadoop.hdfs;
+import static org.apache.hadoop.fs.StreamCapabilities.StreamCapability.HFLUSH;
+import static org.apache.hadoop.fs.StreamCapabilities.StreamCapability.HSYNC;
+
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InterruptedIOException;
@@ -34,6 +37,7 @@ import org.apache.hadoop.fs.FSOutputSummer;
import org.apache.hadoop.fs.FileAlreadyExistsException;
import org.apache.hadoop.fs.FileEncryptionInfo;
import org.apache.hadoop.fs.ParentNotDirectoryException;
+import org.apache.hadoop.fs.StreamCapabilities;
import org.apache.hadoop.fs.Syncable;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
@@ -90,7 +94,7 @@ import com.google.common.base.Preconditions;
****************************************************************/
@InterfaceAudience.Private
public class DFSOutputStream extends FSOutputSummer
- implements Syncable, CanSetDropBehind {
+ implements Syncable, CanSetDropBehind, StreamCapabilities {
static final Logger LOG = LoggerFactory.getLogger(DFSOutputStream.class);
/**
* Number of times to retry creating a file when there are transient
@@ -546,6 +550,15 @@ public class DFSOutputStream extends FSOutputSummer
}
}
+ @Override
+ public boolean hasCapability(String capability) {
+ if (capability.equalsIgnoreCase(HSYNC.getValue()) ||
+ capability.equalsIgnoreCase((HFLUSH.getValue()))) {
+ return true;
+ }
+ return false;
+ }
+
/**
* Flushes out to all replicas of the block. The data is in the buffers
* of the DNs but not necessarily in the DN's OS buffers.
http://git-wip-us.apache.org/repos/asf/hadoop/blob/54fd0e44/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSStripedOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSStripedOutputStream.java b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSStripedOutputStream.java
index 0fdae8c..2aa9e98 100644
--- a/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSStripedOutputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSStripedOutputStream.java
@@ -46,6 +46,7 @@ import com.google.common.annotations.VisibleForTesting;
import org.apache.hadoop.HadoopIllegalArgumentException;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.fs.CreateFlag;
+import org.apache.hadoop.fs.StreamCapabilities;
import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
import org.apache.hadoop.hdfs.protocol.ClientProtocol;
import org.apache.hadoop.hdfs.protocol.DatanodeID;
@@ -77,8 +78,8 @@ import org.apache.htrace.core.TraceScope;
* Each stripe contains a sequence of cells.
*/
@InterfaceAudience.Private
-public class DFSStripedOutputStream extends DFSOutputStream {
-
+public class DFSStripedOutputStream extends DFSOutputStream
+ implements StreamCapabilities {
private static final ByteBufferPool BUFFER_POOL = new ElasticByteBufferPool();
static class MultipleBlockingQueue<T> {
@@ -810,6 +811,12 @@ public class DFSStripedOutputStream extends DFSOutputStream {
}
@Override
+ public boolean hasCapability(String capability) {
+ // StreamCapabilities like hsync / hflush are not supported yet.
+ return false;
+ }
+
+ @Override
public void hflush() {
// not supported yet
}
http://git-wip-us.apache.org/repos/asf/hadoop/blob/54fd0e44/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSOutputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSOutputStream.java
index 52e3bb4..f281a3b 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSOutputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSOutputStream.java
@@ -17,9 +17,11 @@
*/
package org.apache.hadoop.hdfs;
+import java.io.ByteArrayInputStream;
import java.io.DataOutputStream;
import java.io.File;
import java.io.IOException;
+import java.io.InputStream;
import java.lang.reflect.Field;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
@@ -32,8 +34,10 @@ import java.util.Random;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CreateFlag;
import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsTracer;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.StreamCapabilities.StreamCapability;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.DataStreamer.LastExceptionInStreamer;
import org.apache.hadoop.hdfs.client.impl.DfsClientConf;
@@ -48,6 +52,7 @@ import org.apache.hadoop.hdfs.server.blockmanagement.BlockManager;
import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeDescriptor;
import org.apache.hadoop.hdfs.server.blockmanagement.DatanodeManager;
import org.apache.hadoop.hdfs.server.protocol.DatanodeStorage;
+import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.test.GenericTestUtils;
import org.apache.hadoop.test.PathUtils;
import org.apache.htrace.core.SpanId;
@@ -55,6 +60,8 @@ import org.junit.AfterClass;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
+
+import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.anyBoolean;
import static org.mockito.Matchers.anyLong;
import org.mockito.Mockito;
@@ -346,6 +353,24 @@ public class TestDFSOutputStream {
verify(spyClient, times(1)).endFileLease(anyLong());
}
+ @Test
+ public void testStreamFlush() throws Exception {
+ FileSystem fs = cluster.getFileSystem();
+ FSDataOutputStream os = fs.create(new Path("/normal-file"));
+ // Verify output stream supports hsync() and hflush().
+ assertTrue("DFSOutputStream should support hflush()!",
+ os.hasCapability(StreamCapability.HFLUSH.getValue()));
+ assertTrue("DFSOutputStream should support hsync()!",
+ os.hasCapability(StreamCapability.HSYNC.getValue()));
+ byte[] bytes = new byte[1024];
+ InputStream is = new ByteArrayInputStream(bytes);
+ IOUtils.copyBytes(is, os, bytes.length);
+ os.hflush();
+ IOUtils.copyBytes(is, os, bytes.length);
+ os.hsync();
+ os.close();
+ }
+
@AfterClass
public static void tearDown() {
if (cluster != null) {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/54fd0e44/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSStripedOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSStripedOutputStream.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSStripedOutputStream.java
index 70309c9..c0cfea2 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSStripedOutputStream.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSStripedOutputStream.java
@@ -17,7 +17,7 @@
*/
package org.apache.hadoop.hdfs;
-import static org.apache.hadoop.fs.contract.ContractTestUtils.fail;
+import static org.junit.Assert.assertFalse;
import java.io.ByteArrayInputStream;
import java.io.IOException;
@@ -29,6 +29,7 @@ import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.StreamCapabilities.StreamCapability;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
import org.apache.hadoop.io.IOUtils;
@@ -195,14 +196,19 @@ public class TestDFSStripedOutputStream {
public void testStreamFlush() throws Exception {
final byte[] bytes = StripedFileTestUtil.generateBytes(blockSize *
dataBlocks * 3 + cellSize * dataBlocks + cellSize + 123);
- try (FSDataOutputStream os = fs.create(new Path("/ec-file-1"))) {
- InputStream is = new ByteArrayInputStream(bytes);
- IOUtils.copyBytes(is, os, bytes.length);
- os.hflush();
- os.hsync();
- } catch (Exception e) {
- fail("hflush()/hsync() on striped file output stream failed!", e);
- }
+ FSDataOutputStream os = fs.create(new Path("/ec-file-1"));
+ assertFalse("DFSStripedOutputStream should not have hflush() " +
+ "capability yet!", os.hasCapability(
+ StreamCapability.HFLUSH.getValue()));
+ assertFalse("DFSStripedOutputStream should not have hsync() " +
+ "capability yet!", os.hasCapability(
+ StreamCapability.HSYNC.getValue()));
+ InputStream is = new ByteArrayInputStream(bytes);
+ IOUtils.copyBytes(is, os, bytes.length);
+ os.hflush();
+ IOUtils.copyBytes(is, os, bytes.length);
+ os.hsync();
+ os.close();
}
private void testOneFile(String src, int writeBytes) throws Exception {
http://git-wip-us.apache.org/repos/asf/hadoop/blob/54fd0e44/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SyncableDataOutputStream.java
----------------------------------------------------------------------
diff --git a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SyncableDataOutputStream.java b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SyncableDataOutputStream.java
index 9bec7a5..a52fdb7 100644
--- a/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SyncableDataOutputStream.java
+++ b/hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/SyncableDataOutputStream.java
@@ -22,6 +22,7 @@ import java.io.DataOutputStream;
import java.io.IOException;
import java.io.OutputStream;
+import org.apache.hadoop.fs.StreamCapabilities;
import org.apache.hadoop.fs.Syncable;
/**
@@ -30,13 +31,22 @@ import org.apache.hadoop.fs.Syncable;
* wrapped stream passed in to the constructor. This is required
* for HBase when wrapping a PageBlobOutputStream used as a write-ahead log.
*/
-public class SyncableDataOutputStream extends DataOutputStream implements Syncable {
+public class SyncableDataOutputStream extends DataOutputStream
+ implements Syncable, StreamCapabilities {
public SyncableDataOutputStream(OutputStream out) {
super(out);
}
@Override
+ public boolean hasCapability(String capability) {
+ if (out instanceof StreamCapabilities) {
+ return ((StreamCapabilities) out).hasCapability(capability);
+ }
+ return false;
+ }
+
+ @Override
public void hflush() throws IOException {
if (out instanceof Syncable) {
((Syncable) out).hflush();
---------------------------------------------------------------------
To unsubscribe, e-mail: common-commits-unsubscribe@hadoop.apache.org
For additional commands, e-mail: common-commits-help@hadoop.apache.org