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

HDFS-12999. When reach the end of the block group, it may not need to flush all the data packets(flushAllInternals) twice. Contributed by lufei and Fei Hui.

Browse Source 
(cherry picked from commit df622cf4a32ee172ded6c4b3b97a1e49befc4f10)
This commit is contained in:
Ayush Saxena 2019-12-25 11:07:25 +05:30 committed by Wei-Chiu Chuang
parent abee7402e4
commit 354e63b366
1 changed files with 10 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -556,7 +556,7 @@ protected synchronized void writeChunk(byte[] bytes, int offset, int len,
// if this is the end of the block group, end each internal block
if (shouldEndBlockGroup()) {
flushAllInternals();
checkStreamerFailures();
checkStreamerFailures(false);
for (int i = 0; i < numAllBlocks; i++) {
final StripedDataStreamer s = setCurrentStreamer(i);
if (s.isHealthy()) {
@ -567,7 +567,7 @@ protected synchronized void writeChunk(byte[] bytes, int offset, int len,
}
} else {
// check failure state for all the streamers. Bump GS if necessary
checkStreamerFailures();
checkStreamerFailures(true);
}
}
setCurrentStreamer(next);
@ -623,15 +623,18 @@ private Set<StripedDataStreamer> markExternalErrorOnStreamers() {
* written a full stripe (i.e., enqueue all packets for a full stripe), or
* when we're closing the outputstream.
*/
private void checkStreamerFailures() throws IOException {
private void checkStreamerFailures(boolean isNeedFlushAllPackets)
throws IOException {
Set<StripedDataStreamer> newFailed = checkStreamers();
if (newFailed.size() == 0) {
return;
}
// for healthy streamers, wait till all of them have fetched the new block
// and flushed out all the enqueued packets.
flushAllInternals();
if (isNeedFlushAllPackets) {
// for healthy streamers, wait till all of them have fetched the new block
// and flushed out all the enqueued packets.
flushAllInternals();
}
// recheck failed streamers again after the flush
newFailed = checkStreamers();
while (newFailed.size() > 0) {
@ -1188,7 +1191,7 @@ protected synchronized void closeImpl() throws IOException {
// flush all the data packets
flushAllInternals();
// check failures
checkStreamerFailures();
checkStreamerFailures(false);
for (int i = 0; i < numAllBlocks; i++) {
final StripedDataStreamer s = setCurrentStreamer(i);