TestDFSStripedOutputStreamWithFailure.java
/**
* 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;
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.client.HdfsClientConfigKeys;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.HdfsConstants.DatanodeReportType;
import org.apache.hadoop.hdfs.server.datanode.DataNode;
import org.apache.hadoop.test.LambdaTestUtils;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.Timeout;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;
/**
* Test striped file write operation with data node failures with fixed
* parameter test cases.
*/
public class TestDFSStripedOutputStreamWithFailure extends
TestDFSStripedOutputStreamWithFailureBase{
public static final Logger LOG = LoggerFactory.getLogger(
TestDFSStripedOutputStreamWithFailure.class);
@Test
@Timeout(value = 300)
public void testMultipleDatanodeFailure56() throws Exception {
runTestWithMultipleFailure(getLength(56));
}
/**
* Randomly pick a length and run tests with multiple data failures.
* TODO: enable this later
*/
//@Test(timeout=240000)
public void testMultipleDatanodeFailureRandomLength() throws Exception {
int lenIndex = RANDOM.nextInt(lengths.size());
LOG.info("run testMultipleDatanodeFailureRandomLength with length index: "
+ lenIndex);
runTestWithMultipleFailure(getLength(lenIndex));
}
@Test
@Timeout(value = 240)
public void testBlockTokenExpired() throws Exception {
// Make sure killPos is greater than the length of one stripe
final int length = dataBlocks * cellSize * 3;
final HdfsConfiguration conf = newHdfsConfiguration();
conf.setBoolean(DFSConfigKeys.DFS_BLOCK_ACCESS_TOKEN_ENABLE_KEY, true);
conf.setInt(
CommonConfigurationKeysPublic.IPC_CLIENT_CONNECT_MAX_RETRIES_KEY, 0);
// Set short retry timeouts so this test runs faster
conf.setInt(HdfsClientConfigKeys.Retry.WINDOW_BASE_KEY, 10);
for (int dn = 0; dn < dataBlocks + parityBlocks; dn += 2) {
try {
setup(conf);
runTest(length, new int[]{length / 2}, new int[]{dn}, true);
} catch (Exception e) {
LOG.error("failed, dn=" + dn + ", length=" + length);
throw e;
} finally {
tearDown();
}
}
}
@Test
@Timeout(value = 90)
public void testAddBlockWhenNoSufficientDataBlockNumOfNodes()
throws Exception {
HdfsConfiguration conf = new HdfsConfiguration();
conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, blockSize);
try {
setup(conf);
ArrayList<DataNode> dataNodes = cluster.getDataNodes();
// shutdown few datanodes to avoid getting sufficient data blocks number
// of datanodes
int numDatanodes = dataNodes.size();
while (numDatanodes >= dataBlocks) {
cluster.stopDataNode(0);
numDatanodes--;
}
cluster.restartNameNodes();
cluster.triggerHeartbeats();
DatanodeInfo[] info = dfs.getClient().datanodeReport(
DatanodeReportType.LIVE);
assertEquals(numDatanodes, info.length, "Mismatches number of live Dns");
final Path dirFile = new Path(dir, "ecfile");
LambdaTestUtils.intercept(
IOException.class,
"File " + dirFile + " could only be written to " +
numDatanodes + " of the " + dataBlocks + " required nodes for " +
ecPolicy.getName(),
() -> {
try (FSDataOutputStream out = dfs.create(dirFile, true)) {
out.write("something".getBytes());
out.flush();
}
return 0;
});
} finally {
tearDown();
}
}
private void testCloseWithExceptionsInStreamer(
int numFailures, boolean shouldFail) throws Exception {
assertTrue(numFailures <=
ecPolicy.getNumDataUnits() + ecPolicy.getNumParityUnits());
final Path dirFile = new Path(dir, "ecfile-" + numFailures);
try (FSDataOutputStream out = dfs.create(dirFile, true)) {
out.write("idempotent close".getBytes());
// Expect to raise IOE on the first close call, but any following
// close() should be no-op.
LambdaTestUtils.intercept(IOException.class,
out::close);
assertTrue(out.getWrappedStream() instanceof DFSStripedOutputStream);
DFSStripedOutputStream stripedOut =
(DFSStripedOutputStream) out.getWrappedStream();
for (int i = 0; i < numFailures; i++) {
// Only inject 1 stream failure.
stripedOut.getStripedDataStreamer(i).getLastException().set(
new IOException("injected failure")
);
}
if (shouldFail) {
LambdaTestUtils.intercept(IOException.class, out::close);
}
// Close multiple times. All the following close() should have no
// side-effect.
out.close();
}
}
// HDFS-12612
@Test
public void testIdempotentCloseWithFailedStreams() throws Exception {
HdfsConfiguration conf = new HdfsConfiguration();
conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, blockSize);
try {
setup(conf);
// shutdown few datanodes to avoid getting sufficient data blocks number
// of datanodes.
while (cluster.getDataNodes().size() >= dataBlocks) {
cluster.stopDataNode(0);
}
cluster.restartNameNodes();
cluster.triggerHeartbeats();
testCloseWithExceptionsInStreamer(1, false);
testCloseWithExceptionsInStreamer(ecPolicy.getNumParityUnits(), false);
testCloseWithExceptionsInStreamer(ecPolicy.getNumParityUnits() + 1, true);
testCloseWithExceptionsInStreamer(ecPolicy.getNumDataUnits(), true);
} finally {
tearDown();
}
}
@Test
public void testCloseAfterAbort() throws Exception {
HdfsConfiguration conf = new HdfsConfiguration();
conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, blockSize);
try {
setup(conf);
final Path dirFile = new Path(dir, "ecfile");
FSDataOutputStream out = dfs.create(dirFile, true);
assertTrue(out.getWrappedStream() instanceof DFSStripedOutputStream);
DFSStripedOutputStream stripedOut =
(DFSStripedOutputStream) out.getWrappedStream();
stripedOut.abort();
LambdaTestUtils.intercept(IOException.class,
"Lease timeout", stripedOut::close);
} finally {
tearDown();
}
}
@Test
@Timeout(value = 90)
public void testAddBlockWhenNoSufficientParityNumOfNodes()
throws IOException {
HdfsConfiguration conf = new HdfsConfiguration();
conf.setLong(DFSConfigKeys.DFS_BLOCK_SIZE_KEY, blockSize);
try {
setup(conf);
ArrayList<DataNode> dataNodes = cluster.getDataNodes();
// shutdown few data nodes to avoid writing parity blocks
int killDns = (parityBlocks - 1);
int numDatanodes = dataNodes.size() - killDns;
for (int i = 0; i < killDns; i++) {
cluster.stopDataNode(i);
}
cluster.restartNameNodes();
cluster.triggerHeartbeats();
DatanodeInfo[] info = dfs.getClient().datanodeReport(
DatanodeReportType.LIVE);
assertEquals(numDatanodes, info.length, "Mismatches number of live Dns");
Path srcPath = new Path(dir, "testAddBlockWhenNoSufficientParityNodes");
int fileLength = cellSize - 1000;
final byte[] expected = StripedFileTestUtil.generateBytes(fileLength);
DFSTestUtil.writeFile(dfs, srcPath, new String(expected));
LOG.info("writing finished. Seek and read the file to verify.");
StripedFileTestUtil.verifySeek(dfs, srcPath, fileLength, ecPolicy,
blockGroupSize);
} finally {
tearDown();
}
}
/**
* When the two DataNodes with partial data blocks fail.
*/
@Test
public void testCloseWithExceptionsInStreamer() throws Exception {
final HdfsConfiguration conf = newHdfsConfiguration();
final int[] fileLengths = {
// Full stripe then partial on cell boundary
cellSize * (dataBlocks * 2 - 2),
// Full stripe and a partial on non-cell boundary
(cellSize * dataBlocks) + 123,
};
// select the two DNs with partial block to kill
int[] dnIndex = null;
if (parityBlocks > 1) {
dnIndex = new int[] {dataBlocks - 2, dataBlocks - 1};
} else {
dnIndex = new int[] {dataBlocks - 1};
}
for (int length : fileLengths) {
final int[] killPos = getKillPositions(length, dnIndex.length);
try {
LOG.info("runTestWithMultipleFailure2: length==" + length + ", killPos="
+ Arrays.toString(killPos) + ", dnIndex="
+ Arrays.toString(dnIndex));
setup(conf);
runTest(length, killPos, dnIndex, false);
} catch (Throwable e) {
final String err = "failed, killPos=" + Arrays.toString(killPos)
+ ", dnIndex=" + Arrays.toString(dnIndex) + ", length=" + length;
LOG.error(err);
throw e;
} finally {
tearDown();
}
}
}
/**
* Test writing very short EC files with many failures.
*/
@Test
public void runTestWithShortStripe() throws Exception {
final HdfsConfiguration conf = newHdfsConfiguration();
// Write a file with a 1 cell partial stripe
final int length = cellSize - 123;
// Kill all but one DN
final int[] dnIndex = new int[dataBlocks + parityBlocks - 1];
for (int i = 0; i < dnIndex.length; i++) {
dnIndex[i] = i;
}
final int[] killPos = getKillPositions(length, dnIndex.length);
try {
LOG.info("runTestWithShortStripe: length==" + length + ", killPos="
+ Arrays.toString(killPos) + ", dnIndex="
+ Arrays.toString(dnIndex));
setup(conf);
runTest(length, killPos, dnIndex, false);
} catch (Throwable e) {
final String err = "failed, killPos=" + Arrays.toString(killPos)
+ ", dnIndex=" + Arrays.toString(dnIndex) + ", length=" + length;
LOG.error(err);
throw e;
} finally {
tearDown();
}
}
}