Package org.apache.hadoop.hbase.io.asyncfs

Class FanOutOneBlockAsyncDFSOutput

java.lang.Object

org.apache.hadoop.hbase.io.asyncfs.FanOutOneBlockAsyncDFSOutput

All Implemented Interfaces:

Closeable, AutoCloseable, AsyncFSOutput

@Private
public class FanOutOneBlockAsyncDFSOutput
extends Object
implements AsyncFSOutput

An asynchronous HDFS output stream implementation which fans out data to datanode and only supports writing file with only one block.

Use the createOutput method in FanOutOneBlockAsyncDFSOutputHelper to create. The main usage of this class is implementing WAL, so we only expose a little HDFS configurations in the method. And we place it here under io package because we want to make it independent of WAL implementation thus easier to move it to HDFS project finally.

Note that, although we support pipelined flush, i.e, write new data and then flush before the previous flush succeeds, the implementation is not thread safe, so you should not call its methods concurrently.

Advantages compare to DFSOutputStream:

1. The fan out mechanism. This will reduce the latency.
2. Fail-fast when connection to datanode error. The WAL implementation could open new writer ASAP.
3. We could benefit from netty's ByteBuf management mechanism.

Nested Class Summary

Nested Classes

Modifier and Type	Class	Description
private final class	FanOutOneBlockAsyncDFSOutput.AckHandler
private static final class	FanOutOneBlockAsyncDFSOutput.Callback
private static enum	FanOutOneBlockAsyncDFSOutput.State

Field Summary

Fields

Modifier and Type	Field	Description
private long	ackedBlockLength
private final org.apache.hbase.thirdparty.io.netty.buffer.ByteBufAllocator	alloc
private final org.apache.hadoop.hdfs.protocol.ExtendedBlock	block
private org.apache.hbase.thirdparty.io.netty.buffer.ByteBuf	buf
private final org.apache.hadoop.hdfs.DFSClient	client
private final String	clientName
private final org.apache.hadoop.conf.Configuration	conf
private final Map<org.apache.hbase.thirdparty.io.netty.channel.Channel,org.apache.hadoop.hdfs.protocol.DatanodeInfo>	datanodeInfoMap
private final org.apache.hadoop.hdfs.DistributedFileSystem	dfs
private final DummyDFSOutputStream	dummyStream
private final org.apache.hadoop.crypto.Encryptor	encryptor
private final org.apache.hadoop.hdfs.protocol.DatanodeInfo[]	locations
private static final int	MAX_DATA_LEN
private final int	maxDataLen
private final org.apache.hadoop.hdfs.protocol.ClientProtocol	namenode
private long	nextPacketOffsetInBlock
private long	nextPacketSeqno
private final SendBufSizePredictor	sendBufSizePRedictor
private final String	src
private final org.apache.hadoop.hdfs.protocol.HdfsFileStatus	stat
private FanOutOneBlockAsyncDFSOutput.State	state
private final StreamSlowMonitor	streamSlowMonitor
private final org.apache.hadoop.util.DataChecksum	summer
private int	trailingPartialChunkLength
private final ConcurrentLinkedDeque<FanOutOneBlockAsyncDFSOutput.Callback>	waitingAckQueue

Constructor Summary

Constructors

Constructor

Description

FanOutOneBlockAsyncDFSOutput(org.apache.hadoop.conf.Configuration conf, org.apache.hadoop.hdfs.DistributedFileSystem dfs, org.apache.hadoop.hdfs.DFSClient client, org.apache.hadoop.hdfs.protocol.ClientProtocol namenode, String clientName, String src, org.apache.hadoop.hdfs.protocol.HdfsFileStatus stat, EnumSet<org.apache.hadoop.fs.CreateFlag> createFlags, org.apache.hadoop.hdfs.protocol.LocatedBlock locatedBlock, org.apache.hadoop.crypto.Encryptor encryptor, Map<org.apache.hbase.thirdparty.io.netty.channel.Channel,org.apache.hadoop.hdfs.protocol.DatanodeInfo> datanodeInfoMap, org.apache.hadoop.util.DataChecksum summer, org.apache.hbase.thirdparty.io.netty.buffer.ByteBufAllocator alloc, StreamSlowMonitor streamSlowMonitor)

Method Summary

Modifier and Type	Method	Description
int	buffered()	Return the current size of buffered data.
void	close()	End the current block and complete file at namenode.
private void	closeDataNodeChannelsAndAwait()
private void	completed(org.apache.hbase.thirdparty.io.netty.channel.Channel channel)
private void	endBlock()
private void	failed(org.apache.hbase.thirdparty.io.netty.channel.Channel channel, Supplier<Throwable> errorSupplier)
private void	failWaitingAckQueue(org.apache.hbase.thirdparty.io.netty.channel.Channel channel, Supplier<Throwable> errorSupplier)
CompletableFuture<Long>	flush(boolean syncBlock)	Flush the buffer out to datanodes.
private void	flush0(CompletableFuture<Long> future, boolean syncBlock)
private void	flushBuffer(CompletableFuture<Long> future, org.apache.hbase.thirdparty.io.netty.buffer.ByteBuf dataBuf, long nextPacketOffsetInBlock, boolean syncBlock)
(package private) org.apache.hadoop.hdfs.DFSClient	getClient()
(package private) Map<org.apache.hbase.thirdparty.io.netty.channel.Channel,org.apache.hadoop.hdfs.protocol.DatanodeInfo>	getDatanodeInfoMap()
(package private) DummyDFSOutputStream	getDummyStream()
org.apache.hadoop.hdfs.protocol.DatanodeInfo[]	getPipeline()	Return current pipeline.
(package private) org.apache.hadoop.hdfs.protocol.HdfsFileStatus	getStat()
long	getSyncedLength()	Returns byteSize success synced to underlying filesystem.
boolean	isBroken()	Whether the stream is broken.
(package private) boolean	isClosed()
void	recoverAndClose(CancelableProgressable reporter)	The close method when error occurred.
private void	setupReceiver(int timeoutMs)
void	write(byte[] b)	Just call write(b, 0, b.length).
void	write(byte[] b, int off, int len)	Copy the data into the buffer.
void	write(ByteBuffer bb)	Copy the data in the given bb into the buffer.
void	writeInt(int i)	Write an int to the buffer.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

MAX_DATA_LEN

private static final int MAX_DATA_LEN

See Also:

• Constant Field Values

conf

private final org.apache.hadoop.conf.Configuration conf

dfs

private final org.apache.hadoop.hdfs.DistributedFileSystem dfs

client

private final org.apache.hadoop.hdfs.DFSClient client

namenode

private final org.apache.hadoop.hdfs.protocol.ClientProtocol namenode

clientName

private final String clientName

src

private final String src

stat

private final org.apache.hadoop.hdfs.protocol.HdfsFileStatus stat

block

private final org.apache.hadoop.hdfs.protocol.ExtendedBlock block

locations

private final org.apache.hadoop.hdfs.protocol.DatanodeInfo[] locations

encryptor

private final org.apache.hadoop.crypto.Encryptor encryptor

datanodeInfoMap

private final Map<org.apache.hbase.thirdparty.io.netty.channel.Channel,org.apache.hadoop.hdfs.protocol.DatanodeInfo> datanodeInfoMap

summer

private final org.apache.hadoop.util.DataChecksum summer

maxDataLen

private final int maxDataLen

alloc

private final org.apache.hbase.thirdparty.io.netty.buffer.ByteBufAllocator alloc

dummyStream

private final DummyDFSOutputStream dummyStream

waitingAckQueue

private final ConcurrentLinkedDeque<FanOutOneBlockAsyncDFSOutput.Callback> waitingAckQueue

ackedBlockLength

private volatile long ackedBlockLength

nextPacketOffsetInBlock

private long nextPacketOffsetInBlock

trailingPartialChunkLength

private int trailingPartialChunkLength

nextPacketSeqno

private long nextPacketSeqno

buf

private org.apache.hbase.thirdparty.io.netty.buffer.ByteBuf buf

sendBufSizePRedictor

private final SendBufSizePredictor sendBufSizePRedictor

state

private volatile FanOutOneBlockAsyncDFSOutput.State state

streamSlowMonitor

private final StreamSlowMonitor streamSlowMonitor

Constructor Details

FanOutOneBlockAsyncDFSOutput

FanOutOneBlockAsyncDFSOutput(org.apache.hadoop.conf.Configuration conf,
 org.apache.hadoop.hdfs.DistributedFileSystem dfs,
 org.apache.hadoop.hdfs.DFSClient client,
 org.apache.hadoop.hdfs.protocol.ClientProtocol namenode,
 String clientName,
 String src,
 org.apache.hadoop.hdfs.protocol.HdfsFileStatus stat,
 EnumSet<org.apache.hadoop.fs.CreateFlag> createFlags,
 org.apache.hadoop.hdfs.protocol.LocatedBlock locatedBlock,
 org.apache.hadoop.crypto.Encryptor encryptor,
 Map<org.apache.hbase.thirdparty.io.netty.channel.Channel,org.apache.hadoop.hdfs.protocol.DatanodeInfo> datanodeInfoMap,
 org.apache.hadoop.util.DataChecksum summer,
 org.apache.hbase.thirdparty.io.netty.buffer.ByteBufAllocator alloc,
 StreamSlowMonitor streamSlowMonitor)

Method Details

completed

private void completed(org.apache.hbase.thirdparty.io.netty.channel.Channel channel)

failed

private void failed(org.apache.hbase.thirdparty.io.netty.channel.Channel channel,
 Supplier<Throwable> errorSupplier)

failWaitingAckQueue

private void failWaitingAckQueue(org.apache.hbase.thirdparty.io.netty.channel.Channel channel,
 Supplier<Throwable> errorSupplier)

setupReceiver

private void setupReceiver(int timeoutMs)

writeInt

public void writeInt(int i)

Description copied from interface: AsyncFSOutput

Write an int to the buffer.

Specified by:

writeInt in interface AsyncFSOutput

write

public void write(ByteBuffer bb)

Description copied from interface: AsyncFSOutput

Copy the data in the given bb into the buffer.

Specified by:

write in interface AsyncFSOutput

write

public void write(byte[] b)

Description copied from interface: AsyncFSOutput

Just call write(b, 0, b.length).

Specified by:

write in interface AsyncFSOutput

See Also:

• AsyncFSOutput.write(byte[], int, int)

write

public void write(byte[] b,
 int off,
 int len)

Description copied from interface: AsyncFSOutput

Copy the data into the buffer. Note that you need to call AsyncFSOutput.flush(boolean) to flush the buffer manually.

Specified by:

write in interface AsyncFSOutput

buffered

public int buffered()

Description copied from interface: AsyncFSOutput

Return the current size of buffered data.

Specified by:

buffered in interface AsyncFSOutput

getPipeline

public org.apache.hadoop.hdfs.protocol.DatanodeInfo[] getPipeline()

Description copied from interface: AsyncFSOutput

Return current pipeline. Empty array if no pipeline.

Specified by:

getPipeline in interface AsyncFSOutput

flushBuffer

private void flushBuffer(CompletableFuture<Long> future,
 org.apache.hbase.thirdparty.io.netty.buffer.ByteBuf dataBuf,
 long nextPacketOffsetInBlock,
 boolean syncBlock)

flush0

private void flush0(CompletableFuture<Long> future,
 boolean syncBlock)

flush

public CompletableFuture<Long> flush(boolean syncBlock)

Flush the buffer out to datanodes.

Specified by:

flush in interface AsyncFSOutput

Parameters:

syncBlock - will call hsync if true, otherwise hflush.

Returns:

A CompletableFuture that hold the acked length after flushing.

endBlock

private void endBlock()
               throws IOException

Throws:

IOException

closeDataNodeChannelsAndAwait

private void closeDataNodeChannelsAndAwait()

recoverAndClose

public void recoverAndClose(CancelableProgressable reporter)
                     throws IOException

The close method when error occurred. Now we just call recoverFileLease.

Specified by:

recoverAndClose in interface AsyncFSOutput

Throws:

IOException

close

public void close()
           throws IOException

End the current block and complete file at namenode. You should call recoverAndClose(CancelableProgressable) if this method throws an exception.

Specified by:

close in interface AsyncFSOutput

Specified by:

close in interface AutoCloseable

Specified by:

close in interface Closeable

Throws:

IOException

isBroken

public boolean isBroken()

Description copied from interface: AsyncFSOutput

Whether the stream is broken.

Specified by:

isBroken in interface AsyncFSOutput

getSyncedLength

public long getSyncedLength()

Description copied from interface: AsyncFSOutput

Returns byteSize success synced to underlying filesystem.

Specified by:

getSyncedLength in interface AsyncFSOutput

getDatanodeInfoMap

Map<org.apache.hbase.thirdparty.io.netty.channel.Channel,org.apache.hadoop.hdfs.protocol.DatanodeInfo> getDatanodeInfoMap()

getClient

org.apache.hadoop.hdfs.DFSClient getClient()

getDummyStream

DummyDFSOutputStream getDummyStream()

isClosed

boolean isClosed()

getStat

org.apache.hadoop.hdfs.protocol.HdfsFileStatus getStat()