HDFS-8233. Fix DFSStripedOutputStream#getCurrentBlockGroupBytes when the last stripe is at the block group boundary. Contributed by Jing Zhao.
This commit is contained in:
parent
fcd54ecce2
commit
eb612b0b70
|
@ -122,3 +122,6 @@
|
||||||
|
|
||||||
HDFS-8136. Client gets and uses EC schema when reads and writes a stripping
|
HDFS-8136. Client gets and uses EC schema when reads and writes a stripping
|
||||||
file. (Kai Sasaki via Kai Zheng)
|
file. (Kai Sasaki via Kai Zheng)
|
||||||
|
|
||||||
|
HDFS-8233. Fix DFSStripedOutputStream#getCurrentBlockGroupBytes when the last
|
||||||
|
stripe is at the block group boundary. (jing9)
|
||||||
|
|
|
@ -36,7 +36,6 @@ import org.apache.hadoop.hdfs.protocol.ECInfo;
|
||||||
import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
|
import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
|
||||||
import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
|
import org.apache.hadoop.hdfs.protocol.HdfsFileStatus;
|
||||||
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
|
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
|
||||||
import org.apache.hadoop.hdfs.util.StripedBlockUtil;
|
|
||||||
import org.apache.hadoop.io.erasurecode.rawcoder.RSRawEncoder;
|
import org.apache.hadoop.io.erasurecode.rawcoder.RSRawEncoder;
|
||||||
import org.apache.hadoop.io.erasurecode.rawcoder.RawErasureEncoder;
|
import org.apache.hadoop.io.erasurecode.rawcoder.RawErasureEncoder;
|
||||||
import org.apache.hadoop.util.DataChecksum;
|
import org.apache.hadoop.util.DataChecksum;
|
||||||
|
@ -278,14 +277,6 @@ public class DFSStripedOutputStream extends DFSOutputStream {
|
||||||
return numDataBlocks * cellSize;
|
return numDataBlocks * cellSize;
|
||||||
}
|
}
|
||||||
|
|
||||||
private long getCurrentBlockGroupBytes() {
|
|
||||||
long sum = 0;
|
|
||||||
for (int i = 0; i < numDataBlocks; i++) {
|
|
||||||
sum += streamers.get(i).getBytesCurBlock();
|
|
||||||
}
|
|
||||||
return sum;
|
|
||||||
}
|
|
||||||
|
|
||||||
private void notSupported(String headMsg)
|
private void notSupported(String headMsg)
|
||||||
throws IOException{
|
throws IOException{
|
||||||
throw new IOException(
|
throw new IOException(
|
||||||
|
@ -347,24 +338,31 @@ public class DFSStripedOutputStream extends DFSOutputStream {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Simply add bytesCurBlock together. Note that this result is not accurately
|
||||||
|
* the size of the block group.
|
||||||
|
*/
|
||||||
|
private long getCurrentSumBytes() {
|
||||||
|
long sum = 0;
|
||||||
|
for (int i = 0; i < numDataBlocks; i++) {
|
||||||
|
sum += streamers.get(i).getBytesCurBlock();
|
||||||
|
}
|
||||||
|
return sum;
|
||||||
|
}
|
||||||
|
|
||||||
private void writeParityCellsForLastStripe() throws IOException {
|
private void writeParityCellsForLastStripe() throws IOException {
|
||||||
final long currentBlockGroupBytes = getCurrentBlockGroupBytes();
|
final long currentBlockGroupBytes = getCurrentSumBytes();
|
||||||
long parityBlkSize = StripedBlockUtil.getInternalBlockLength(
|
if (currentBlockGroupBytes % stripeDataSize() == 0) {
|
||||||
currentBlockGroupBytes, cellSize, numDataBlocks,
|
|
||||||
numDataBlocks + 1);
|
|
||||||
if (parityBlkSize == 0 || currentBlockGroupBytes % stripeDataSize() == 0) {
|
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
int parityCellSize = parityBlkSize % cellSize == 0 ? cellSize :
|
long firstCellSize = getLeadingStreamer().getBytesCurBlock() % cellSize;
|
||||||
(int) (parityBlkSize % cellSize);
|
long parityCellSize = firstCellSize > 0 && firstCellSize < cellSize ?
|
||||||
|
firstCellSize : cellSize;
|
||||||
|
|
||||||
for (int i = 0; i < numAllBlocks; i++) {
|
for (int i = 0; i < numAllBlocks; i++) {
|
||||||
long internalBlkLen = StripedBlockUtil.getInternalBlockLength(
|
|
||||||
currentBlockGroupBytes, cellSize, numDataBlocks, i);
|
|
||||||
// Pad zero bytes to make all cells exactly the size of parityCellSize
|
// Pad zero bytes to make all cells exactly the size of parityCellSize
|
||||||
// If internal block is smaller than parity block, pad zero bytes.
|
// If internal block is smaller than parity block, pad zero bytes.
|
||||||
// Also pad zero bytes to all parity cells
|
// Also pad zero bytes to all parity cells
|
||||||
if (internalBlkLen < parityBlkSize || i >= numDataBlocks) {
|
|
||||||
int position = cellBuffers[i].position();
|
int position = cellBuffers[i].position();
|
||||||
assert position <= parityCellSize : "If an internal block is smaller" +
|
assert position <= parityCellSize : "If an internal block is smaller" +
|
||||||
" than parity block, then its last cell should be small than last" +
|
" than parity block, then its last cell should be small than last" +
|
||||||
|
@ -372,12 +370,11 @@ public class DFSStripedOutputStream extends DFSOutputStream {
|
||||||
for (int j = 0; j < parityCellSize - position; j++) {
|
for (int j = 0; j < parityCellSize - position; j++) {
|
||||||
cellBuffers[i].put((byte) 0);
|
cellBuffers[i].put((byte) 0);
|
||||||
}
|
}
|
||||||
}
|
|
||||||
cellBuffers[i].flip();
|
cellBuffers[i].flip();
|
||||||
}
|
}
|
||||||
encode(cellBuffers);
|
encode(cellBuffers);
|
||||||
|
|
||||||
//write parity cells
|
// write parity cells
|
||||||
curIdx = numDataBlocks;
|
curIdx = numDataBlocks;
|
||||||
refreshStreamer();
|
refreshStreamer();
|
||||||
for (int i = numDataBlocks; i < numAllBlocks; i++) {
|
for (int i = numDataBlocks; i < numAllBlocks; i++) {
|
||||||
|
|
|
@ -114,6 +114,12 @@ public class TestDFSStripedOutputStream {
|
||||||
+ cellSize * dataBlocks + 123);
|
+ cellSize * dataBlocks + 123);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testFileLessThanFullBlockGroup() throws IOException {
|
||||||
|
testOneFile("/LessThanFullBlockGroup",
|
||||||
|
cellSize * dataBlocks * (stripesPerBlock - 1) + cellSize);
|
||||||
|
}
|
||||||
|
|
||||||
@Test
|
@Test
|
||||||
public void testFileFullBlockGroup() throws IOException {
|
public void testFileFullBlockGroup() throws IOException {
|
||||||
testOneFile("/FullBlockGroup", blockSize * dataBlocks);
|
testOneFile("/FullBlockGroup", blockSize * dataBlocks);
|
||||||
|
|
Loading…
Reference in New Issue