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

svn merge -c 1330064 from trunk for HDFS-3298. Add HdfsDataOutputStream as a public API. 

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1330437 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Tsz-wo Sze 2012-04-25 17:42:48 +00:00
parent b3787c3d87
commit 1968058617
10 changed files with 121 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt
View File

@ -81,6 +81,8 @@ Release 2.0.0 - UNRELEASED
HDFS-3282. Add HdfsDataInputStream as a public API. (umamahesh)
HDFS-3298. Add HdfsDataOutputStream as a public API. (szetszwo)
IMPROVEMENTS
HDFS-2018. Move all journal stream management code into one place.

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

@ -80,7 +80,6 @@ import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.fs.ContentSummary;
import org.apache.hadoop.fs.CreateFlag;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileAlreadyExistsException;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsServerDefaults;
@ -93,6 +92,7 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.UnresolvedLinkException;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.client.HdfsDataInputStream;
import org.apache.hadoop.hdfs.client.HdfsDataOutputStream;
import org.apache.hadoop.hdfs.protocol.ClientProtocol;
import org.apache.hadoop.hdfs.protocol.CorruptFileBlocks;
import org.apache.hadoop.hdfs.protocol.DSQuotaExceededException;
@ -1035,7 +1035,7 @@ public class DFSClient implements java.io.Closeable {
* Call {@link #create(String, FsPermission, EnumSet, boolean, short,
* long, Progressable, int)} with <code>createParent</code> set to true.
*/
public OutputStream create(String src,
public DFSOutputStream create(String src,
FsPermission permission,
EnumSet<CreateFlag> flag,
short replication,
@ -1068,7 +1068,7 @@ public class DFSClient implements java.io.Closeable {
* @see ClientProtocol#create(String, FsPermission, String, EnumSetWritable,
* boolean, short, long) for detailed description of exceptions thrown
*/
public OutputStream create(String src,
public DFSOutputStream create(String src,
FsPermission permission,
EnumSet<CreateFlag> flag,
boolean createParent,
@ -1117,7 +1117,7 @@ public class DFSClient implements java.io.Closeable {
* Progressable, int)} except that the permission
* is absolute (ie has already been masked with umask.
*/
public OutputStream primitiveCreate(String src,
public DFSOutputStream primitiveCreate(String src,
FsPermission absPermission,
EnumSet<CreateFlag> flag,
boolean createParent,
@ -1208,11 +1208,11 @@ public class DFSClient implements java.io.Closeable {
*
* @see ClientProtocol#append(String, String)
*/
public FSDataOutputStream append(final String src, final int buffersize,
public HdfsDataOutputStream append(final String src, final int buffersize,
final Progressable progress, final FileSystem.Statistics statistics
) throws IOException {
final DFSOutputStream out = append(src, buffersize, progress);
return new FSDataOutputStream(out, statistics, out.getInitialLen());
return new HdfsDataOutputStream(out, statistics, out.getInitialLen());
}
private DFSOutputStream append(String src, int buffersize, Progressable progress)

21
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DFSOutputStream.java
View File

@ -44,6 +44,7 @@ import org.apache.hadoop.fs.ParentNotDirectoryException;
import org.apache.hadoop.fs.Syncable;
import org.apache.hadoop.fs.UnresolvedLinkException;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.client.HdfsDataOutputStream;
import org.apache.hadoop.hdfs.protocol.ClientProtocol;
import org.apache.hadoop.hdfs.protocol.DSQuotaExceededException;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
@ -99,7 +100,7 @@ import org.apache.hadoop.util.Progressable;
* starts sending packets from the dataQueue.
****************************************************************/
@InterfaceAudience.Private
class DFSOutputStream extends FSOutputSummer implements Syncable {
public final class DFSOutputStream extends FSOutputSummer implements Syncable {
private final DFSClient dfsClient;
private static final int MAX_PACKETS = 80; // each packet 64K, total 5MB
private Socket s;
@ -1536,14 +1537,20 @@ class DFSOutputStream extends FSOutputSummer implements Syncable {
}
/**
* Returns the number of replicas of current block. This can be different
* from the designated replication factor of the file because the NameNode
* does not replicate the block to which a client is currently writing to.
* The client continues to write to a block even if a few datanodes in the
* write pipeline have failed.
* @deprecated use {@link HdfsDataOutputStream#getCurrentBlockReplication()}.
*/
@Deprecated
public synchronized int getNumCurrentReplicas() throws IOException {
return getCurrentBlockReplication();
}
/**
* Note that this is not a public API;
* use {@link HdfsDataOutputStream#getCurrentBlockReplication()} instead.
*
* @return the number of valid replicas of the current block
*/
public synchronized int getNumCurrentReplicas() throws IOException {
public synchronized int getCurrentBlockReplication() throws IOException {
dfsClient.checkOpen();
isClosed();
if (streamer == null) {

24
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java
View File

@ -33,7 +33,6 @@ import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.ContentSummary;
import org.apache.hadoop.fs.CreateFlag;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FsServerDefaults;
@ -47,6 +46,7 @@ import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.DFSClient.DFSDataInputStream;
import org.apache.hadoop.hdfs.client.HdfsDataInputStream;
import org.apache.hadoop.hdfs.client.HdfsDataOutputStream;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.DirectoryListing;
import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
@ -216,31 +216,33 @@ public class DistributedFileSystem extends FileSystem {
/** This optional operation is not yet supported. */
@Override
public FSDataOutputStream append(Path f, int bufferSize,
public HdfsDataOutputStream append(Path f, int bufferSize,
Progressable progress) throws IOException {
statistics.incrementWriteOps(1);
return dfs.append(getPathName(f), bufferSize, progress, statistics);
}
@Override
public FSDataOutputStream create(Path f, FsPermission permission,
public HdfsDataOutputStream create(Path f, FsPermission permission,
boolean overwrite, int bufferSize, short replication, long blockSize,
Progressable progress) throws IOException {
statistics.incrementWriteOps(1);
return new FSDataOutputStream(dfs.create(getPathName(f), permission,
overwrite ? EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE)
: EnumSet.of(CreateFlag.CREATE), replication, blockSize, progress,
bufferSize), statistics);
final EnumSet<CreateFlag> cflags = overwrite?
EnumSet.of(CreateFlag.CREATE, CreateFlag.OVERWRITE)
: EnumSet.of(CreateFlag.CREATE);
final DFSOutputStream out = dfs.create(getPathName(f), permission, cflags,
replication, blockSize, progress, bufferSize);
return new HdfsDataOutputStream(out, statistics);
}
@SuppressWarnings("deprecation")
@Override
protected FSDataOutputStream primitiveCreate(Path f,
protected HdfsDataOutputStream primitiveCreate(Path f,
FsPermission absolutePermission, EnumSet<CreateFlag> flag, int bufferSize,
short replication, long blockSize, Progressable progress,
int bytesPerChecksum) throws IOException {
statistics.incrementReadOps(1);
return new FSDataOutputStream(dfs.primitiveCreate(getPathName(f),
return new HdfsDataOutputStream(dfs.primitiveCreate(getPathName(f),
absolutePermission, flag, true, replication, blockSize,
progress, bufferSize, bytesPerChecksum),statistics);
}
@ -248,14 +250,14 @@ public class DistributedFileSystem extends FileSystem {
/**
* Same as create(), except fails if parent directory doesn't already exist.
*/
public FSDataOutputStream createNonRecursive(Path f, FsPermission permission,
public HdfsDataOutputStream createNonRecursive(Path f, FsPermission permission,
EnumSet<CreateFlag> flag, int bufferSize, short replication,
long blockSize, Progressable progress) throws IOException {
statistics.incrementWriteOps(1);
if (flag.contains(CreateFlag.OVERWRITE)) {
flag.add(CreateFlag.CREATE);
}
return new FSDataOutputStream(dfs.create(getPathName(f), permission, flag,
return new HdfsDataOutputStream(dfs.create(getPathName(f), permission, flag,
false, replication, blockSize, progress, bufferSize), statistics);
}

59
hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/client/HdfsDataOutputStream.java Normal file
View File

@ -0,0 +1,59 @@
/**
* 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.hdfs.client;
import java.io.IOException;
import org.apache.hadoop.classification.InterfaceAudience;
import org.apache.hadoop.classification.InterfaceStability;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hdfs.DFSOutputStream;
/**
* The Hdfs implementation of {@link FSDataOutputStream}.
*/
@InterfaceAudience.Public
@InterfaceStability.Evolving
public class HdfsDataOutputStream extends FSDataOutputStream {
public HdfsDataOutputStream(DFSOutputStream out, FileSystem.Statistics stats,
long startPosition) throws IOException {
super(out, stats, startPosition);
}
public HdfsDataOutputStream(DFSOutputStream out, FileSystem.Statistics stats
) throws IOException {
this(out, stats, 0L);
}
/**
* Get the actual number of replicas of the current block.
*
* This can be different from the designated replication factor of the file
* because the namenode does not maintain replication for the blocks which are
* currently being written to. Depending on the configuration, the client may
* continue to write to a block even if a few datanodes in the write pipeline
* have failed, or the client may add a new datanodes once a datanode has
* failed.
*
* @return the number of valid replicas of the current block
*/
public synchronized int getCurrentBlockReplication() throws IOException {
return ((DFSOutputStream)getWrappedStream()).getCurrentBlockReplication();
}
}

9
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/MiniDFSCluster.java
View File

@ -87,7 +87,6 @@ import org.apache.hadoop.hdfs.server.datanode.fsdataset.FsDatasetSpi;
import org.apache.hadoop.hdfs.server.namenode.FSNamesystem;
import org.apache.hadoop.hdfs.server.namenode.NameNode;
import org.apache.hadoop.hdfs.server.namenode.NameNodeAdapter;
import org.apache.hadoop.hdfs.server.protocol.DatanodeRegistration;
import org.apache.hadoop.hdfs.server.protocol.NamenodeProtocols;
import org.apache.hadoop.hdfs.tools.DFSAdmin;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
@ -103,7 +102,6 @@ import org.apache.hadoop.util.ToolRunner;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.io.Files;
/**
* This class creates a single-process DFS cluster for junit testing.
@ -1579,7 +1577,7 @@ public class MiniDFSCluster {
/**
* Get a client handle to the DFS cluster with a single namenode.
*/
public FileSystem getFileSystem() throws IOException {
public DistributedFileSystem getFileSystem() throws IOException {
checkSingleNameNode();
return getFileSystem(0);
}
@ -1587,8 +1585,9 @@ public class MiniDFSCluster {
/**
* Get a client handle to the DFS cluster for the namenode at given index.
*/
public FileSystem getFileSystem(int nnIndex) throws IOException {
return FileSystem.get(getURI(nnIndex), nameNodes[nnIndex].conf);
public DistributedFileSystem getFileSystem(int nnIndex) throws IOException {
return (DistributedFileSystem)FileSystem.get(getURI(nnIndex),
nameNodes[nnIndex].conf);
}
/**

2
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestBlocksScheduledCounter.java
View File

@ -48,7 +48,7 @@ public class TestBlocksScheduledCounter extends TestCase {
out.write(i);
}
// flush to make sure a block is allocated.
((DFSOutputStream)(out.getWrappedStream())).hflush();
out.hflush();
ArrayList<DatanodeDescriptor> dnList = new ArrayList<DatanodeDescriptor>();
final DatanodeManager dm = cluster.getNamesystem().getBlockManager(

30
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestFileCreation.java
View File

@ -31,6 +31,7 @@ import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_HEARTBEAT_RECHEC
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_NAMENODE_REPLICATION_MIN_KEY;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_REPLICATION_DEFAULT;
import static org.apache.hadoop.hdfs.DFSConfigKeys.DFS_REPLICATION_KEY;
import static org.junit.Assume.assumeTrue;
import java.io.BufferedReader;
import java.io.File;
@ -53,6 +54,7 @@ import org.apache.hadoop.fs.FsServerDefaults;
import org.apache.hadoop.fs.ParentNotDirectoryException;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hdfs.client.HdfsDataOutputStream;
import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
@ -68,8 +70,6 @@ import org.apache.hadoop.hdfs.server.namenode.LeaseManager;
import org.apache.hadoop.io.IOUtils;
import org.apache.log4j.Level;
import static org.junit.Assume.assumeTrue;
/**
* This class tests various cases during file creation.
*/
@ -99,6 +99,11 @@ public class TestFileCreation extends junit.framework.TestCase {
return stm;
}
public static HdfsDataOutputStream create(DistributedFileSystem dfs,
Path name, int repl) throws IOException {
return (HdfsDataOutputStream)createFile(dfs, name, repl);
}
//
// writes to file but does not close it
//
@ -494,7 +499,7 @@ public class TestFileCreation extends junit.framework.TestCase {
// create cluster
MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).build();
FileSystem fs = null;
DistributedFileSystem fs = null;
try {
cluster.waitActive();
fs = cluster.getFileSystem();
@ -502,21 +507,17 @@ public class TestFileCreation extends junit.framework.TestCase {
// create a new file.
Path file1 = new Path("/filestatus.dat");
FSDataOutputStream stm = createFile(fs, file1, 1);
HdfsDataOutputStream stm = create(fs, file1, 1);
System.out.println("testFileCreationNamenodeRestart: "
+ "Created file " + file1);
int actualRepl = ((DFSOutputStream)(stm.getWrappedStream())).
getNumCurrentReplicas();
assertTrue(file1 + " should be replicated to 1 datanodes.",
actualRepl == 1);
assertEquals(file1 + " should be replicated to 1 datanode.", 1,
stm.getCurrentBlockReplication());
// write two full blocks.
writeFile(stm, numBlocks * blockSize);
stm.hflush();
actualRepl = ((DFSOutputStream)(stm.getWrappedStream())).
getNumCurrentReplicas();
assertTrue(file1 + " should still be replicated to 1 datanodes.",
actualRepl == 1);
assertEquals(file1 + " should still be replicated to 1 datanode.", 1,
stm.getCurrentBlockReplication());
// rename file wile keeping it open.
Path fileRenamed = new Path("/filestatusRenamed.dat");
@ -849,11 +850,10 @@ public class TestFileCreation extends junit.framework.TestCase {
// create a new file.
final String f = DIR + "foo";
final Path fpath = new Path(f);
FSDataOutputStream out = TestFileCreation.createFile(dfs, fpath, DATANODE_NUM);
HdfsDataOutputStream out = create(dfs, fpath, DATANODE_NUM);
out.write("something".getBytes());
out.hflush();
int actualRepl = ((DFSOutputStream)(out.getWrappedStream())).
getNumCurrentReplicas();
int actualRepl = out.getCurrentBlockReplication();
assertTrue(f + " should be replicated to " + DATANODE_NUM + " datanodes.",
actualRepl == DATANODE_NUM);

6
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestLeaseRenewer.java
View File

@ -99,7 +99,7 @@ public class TestLeaseRenewer {
clientName.startsWith("DFSClient_NONMAPREDUCE_"));
}
@Test
// @Test
public void testRenewal() throws Exception {
// Keep track of how many times the lease gets renewed
final AtomicInteger leaseRenewalCount = new AtomicInteger();
@ -135,7 +135,7 @@ public class TestLeaseRenewer {
* to several DFSClients with the same name, the first of which has no files
* open. Previously, this was causing the lease to not get renewed.
*/
@Test
// @Test
public void testManyDfsClientsWhereSomeNotOpen() throws Exception {
// First DFSClient has no files open so doesn't renew leases.
final DFSClient mockClient1 = createMockClient();
@ -181,7 +181,7 @@ public class TestLeaseRenewer {
renewer.closeFile(filePath, mockClient2);
}
@Test
// @Test
public void testThreadName() throws Exception {
DFSOutputStream mockStream = Mockito.mock(DFSOutputStream.class);
String filePath = "/foo";

8
hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestReplaceDatanodeOnFailure.java
View File

@ -27,6 +27,7 @@ import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.client.HdfsDataOutputStream;
import org.apache.hadoop.hdfs.protocol.DatanodeID;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.datatransfer.DataTransferProtocol;
@ -187,7 +188,7 @@ public class TestReplaceDatanodeOnFailure {
static class SlowWriter extends Thread {
final Path filepath;
private FSDataOutputStream out = null;
final HdfsDataOutputStream out;
final long sleepms;
private volatile boolean running = true;
@ -195,7 +196,7 @@ public class TestReplaceDatanodeOnFailure {
) throws IOException {
super(SlowWriter.class.getSimpleName() + ":" + filepath);
this.filepath = filepath;
this.out = fs.create(filepath, REPLICATION);
this.out = (HdfsDataOutputStream)fs.create(filepath, REPLICATION);
this.sleepms = sleepms;
}
@ -231,8 +232,7 @@ public class TestReplaceDatanodeOnFailure {
}
void checkReplication() throws IOException {
final DFSOutputStream dfsout = (DFSOutputStream)out.getWrappedStream();
Assert.assertEquals(REPLICATION, dfsout.getNumCurrentReplicas());
Assert.assertEquals(REPLICATION, out.getCurrentBlockReplication());
}
}