Apache
/
hadoop
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

HDFS-14027. DFSStripedOutputStream should implement both hsync methods. 

(cherry picked from commit db7e636824)
This commit is contained in:
Xiao Chen 2018-10-29 19:05:52 -07:00
parent 288bc3f1eb
commit 8788489df4
2 changed files with 35 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSStripedOutputStream.java
View File

@ -24,6 +24,7 @@ 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.client.HdfsDataOutputStream.SyncFlag;
import org.apache.hadoop.hdfs.protocol.ClientProtocol;
import org.apache.hadoop.hdfs.protocol.DatanodeID;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
@ -956,11 +957,22 @@ public class DFSStripedOutputStream extends DFSOutputStream
@Override
public void hflush() {
// not supported yet
LOG.debug("DFSStripedOutputStream does not support hflush. "
+ "Caller should check StreamCapabilities before calling.");
}
@Override
public void hsync() {
// not supported yet
LOG.debug("DFSStripedOutputStream does not support hsync. "
+ "Caller should check StreamCapabilities before calling.");
}
@Override
public void hsync(EnumSet<SyncFlag> syncFlags) {
// not supported yet
LOG.debug("DFSStripedOutputStream does not support hsync {}. "
+ "Caller should check StreamCapabilities before calling.", syncFlags);
}
@Override

36
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSStripedOutputStream.java
View File

@ -18,12 +18,14 @@
package org.apache.hadoop.hdfs;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.util.ArrayList;
import java.util.EnumSet;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -31,6 +33,7 @@ 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.client.HdfsDataOutputStream.SyncFlag;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
import org.apache.hadoop.io.IOUtils;
@ -196,19 +199,26 @@ public class TestDFSStripedOutputStream {
public void testStreamFlush() throws Exception {
final byte[] bytes = StripedFileTestUtil.generateBytes(blockSize *
dataBlocks * 3 + cellSize * dataBlocks + cellSize + 123);
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();
try (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()));
try (InputStream is = new ByteArrayInputStream(bytes)) {
IOUtils.copyBytes(is, os, bytes.length);
os.hflush();
IOUtils.copyBytes(is, os, bytes.length);
os.hsync();
IOUtils.copyBytes(is, os, bytes.length);
}
assertTrue("stream is not a DFSStripedOutputStream",
os.getWrappedStream() instanceof DFSStripedOutputStream);
final DFSStripedOutputStream dfssos =
(DFSStripedOutputStream) os.getWrappedStream();
dfssos.hsync(EnumSet.of(SyncFlag.UPDATE_LENGTH));
}
}
private void testOneFile(String src, int writeBytes) throws Exception {