Skip to content

HDFS-17800 Prevent observer node from returning partial block data for erasure coded files #7754

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: trunk
Choose a base branch
from
+117 −5
Open

HDFS-17800 Prevent observer node from returning partial block data for erasure coded files #7754

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 22 additions & 4 deletions ...roject/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2227,11 +2227,20 @@ LocatedBlocks getBlockLocations(String clientMachine, String srcArg,
dir, pc, srcArg, offset, length, true);
inode = res.getIIp().getLastINode();
if (isInSafeMode()) {
int minBlocks = 1;

ErasureCodingPolicy ecPolicy = res.blocks.getErasureCodingPolicy();
for (LocatedBlock b : res.blocks.getLocatedBlocks()) {
if (ecPolicy != null) {
// If the file is erasure coded, we need at least the number of data units of
// blocks available, unless the file is smaller than a full stripe of cells.
long numCells = (b.getBlockSize() - 1) / (long)ecPolicy.getCellSize() + 1;
minBlocks = (int)Math.min((long)ecPolicy.getNumDataUnits(), numCells);
}
// if safemode & no block locations yet then throw safemodeException
if ((b.getLocations() == null) || (b.getLocations().length == 0)) {
if ((b.getLocations() == null) || (b.getLocations().length < minBlocks)) {
SafeModeException se = newSafemodeException(
"Zero blocklocations for " + srcArg);
"Not enough blocklocations for " + srcArg);
if (haEnabled && haContext != null &&
(haContext.getState().getServiceState() == ACTIVE ||
haContext.getState().getServiceState() == OBSERVER)) {
Expand Down Expand Up @@ -9212,9 +9221,18 @@ private void checkBlockLocationsWhenObserver(LocatedBlocks blocks, String src)
}
List<LocatedBlock> locatedBlockList = blocks.getLocatedBlocks();
if (locatedBlockList != null) {
int minBlocks = 1;

ErasureCodingPolicy ecPolicy = blocks.getErasureCodingPolicy();
for (LocatedBlock b : locatedBlockList) {
if (b.getLocations() == null || b.getLocations().length == 0) {
throw new ObserverRetryOnActiveException("Zero blocklocations for " + src);
if (ecPolicy != null) {
// If the file is erasure coded, we need at least the number of data units of
// blocks available, unless the file is smaller than a full stripe of cells.
long numCells = (b.getBlockSize() - 1) / (long)ecPolicy.getCellSize() + 1;
minBlocks = (int)Math.min((long)ecPolicy.getNumDataUnits(), numCells);
}
if (b.getLocations() == null || b.getLocations().length < minBlocks) {
throw new ObserverRetryOnActiveException("Not enough blocklocations for " + src);
}
}
}
Expand Down
96 changes: 95 additions & 1 deletion ...hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/ha/TestObserverNode.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,6 +58,7 @@
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.hadoop.hdfs.protocol.Block;
import org.apache.hadoop.hdfs.protocol.DatanodeInfo;
import org.apache.hadoop.hdfs.protocol.ErasureCodingPolicy;
import org.apache.hadoop.hdfs.protocol.ExtendedBlock;
import org.apache.hadoop.hdfs.protocol.LocatedBlock;
import org.apache.hadoop.hdfs.protocol.LocatedBlocks;
Expand All @@ -69,6 +70,7 @@
import org.apache.hadoop.hdfs.server.namenode.NameNodeRpcServer;
import org.apache.hadoop.hdfs.server.namenode.TestFsck;
import org.apache.hadoop.hdfs.tools.GetGroups;
import org.apache.hadoop.io.erasurecode.ECSchema;
import org.apache.hadoop.ipc.ObserverRetryOnActiveException;
import org.apache.hadoop.ipc.metrics.RpcMetrics;
import org.apache.hadoop.test.GenericTestUtils;
Expand Down Expand Up @@ -437,6 +439,43 @@ public void testObserverNodeSafeModeWithBlockLocations() throws Exception {
dfs.open(testPath).close();
assertSentTo(0);

// Test erasure coded files
ErasureCodingPolicy ecPolicy = new ErasureCodingPolicy(new ECSchema("rs", 3, 2), 1024);

// Fake a small file that only needs 1 block
doAnswer((invocation) -> {
List<LocatedBlock> fakeBlocks = new ArrayList<>();
// Return a single location, which is enough for the small file but not for the large file
ExtendedBlock b = new ExtendedBlock("fake-pool", new Block(12345L, 1, 0));
DatanodeInfo datanodeInfo = new DatanodeInfo.DatanodeInfoBuilder().build();
LocatedBlock fakeBlock = new LocatedBlock(b, new DatanodeInfo[] {datanodeInfo});
fakeBlocks.add(fakeBlock);
return new LocatedBlocks(1, false, fakeBlocks, null, true, null, ecPolicy);
}).when(bmSpy).createLocatedBlocks(Mockito.any(), anyLong(),
anyBoolean(), anyLong(), anyLong(), anyBoolean(), anyBoolean(),
Mockito.any(), Mockito.any());

// Small file should suceed with just the one block
dfs.open(testPath).close();
assertSentTo(2);

// Fake a larger file that needs all 3 data shards
doAnswer((invocation) -> {
List<LocatedBlock> fakeBlocks = new ArrayList<>();
// Return a single location, which is enough for the small file but not for the large file
ExtendedBlock b = new ExtendedBlock("fake-pool", new Block(12345L, 1024 * 3, 0));
DatanodeInfo datanodeInfo = new DatanodeInfo.DatanodeInfoBuilder().build();
LocatedBlock fakeBlock = new LocatedBlock(b, new DatanodeInfo[] {datanodeInfo});
fakeBlocks.add(fakeBlock);
return new LocatedBlocks(1024 * 3, false, fakeBlocks, null, true, null, ecPolicy);
}).when(bmSpy).createLocatedBlocks(Mockito.any(), anyLong(),
anyBoolean(), anyLong(), anyLong(), anyBoolean(), anyBoolean(),
Mockito.any(), Mockito.any());

// Large file should failover to the active
dfs.open(testPath).close();
assertSentTo(0);

Mockito.reset(bmSpy);

// Remove safe mode on observer, request should still go to it.
Expand Down Expand Up @@ -471,7 +510,62 @@ public void testObserverNodeBlockMissingRetry() throws Exception {
anyBoolean(), anyLong(), anyLong(), anyBoolean(), anyBoolean(),
Mockito.any(), Mockito.any());

dfs.open(testPath);
dfs.open(testPath).close();
assertSentTo(0);

dfs.getClient().listPaths("/", new byte[0], true);
assertSentTo(0);

dfs.getClient().getLocatedFileInfo(testPath.toString(), false);
assertSentTo(0);

dfs.getClient().batchedListPaths(new String[]{"/"}, new byte[0], true);
assertSentTo(0);

// Test erasure coded files
ErasureCodingPolicy ecPolicy = new ErasureCodingPolicy(new ECSchema("rs", 3, 2), 1024);

// Fake a small file that only needs 1 block
doAnswer((invocation) -> {
List<LocatedBlock> fakeBlocks = new ArrayList<>();
// Return a single location, which is enough for the small file but not for the large file
ExtendedBlock b = new ExtendedBlock("fake-pool", new Block(12345L, 1, 0));
DatanodeInfo datanodeInfo = new DatanodeInfo.DatanodeInfoBuilder().build();
LocatedBlock fakeBlock = new LocatedBlock(b, new DatanodeInfo[] {datanodeInfo});
fakeBlocks.add(fakeBlock);
return new LocatedBlocks(1, false, fakeBlocks, null, true, null, ecPolicy);
}).when(bmSpy).createLocatedBlocks(Mockito.any(), anyLong(),
anyBoolean(), anyLong(), anyLong(), anyBoolean(), anyBoolean(),
Mockito.any(), Mockito.any());

// The small file should succeed on the observer, while the large file should not

dfs.open(testPath).close();
assertSentTo(2);

dfs.getClient().listPaths("/", new byte[0], true);
assertSentTo(2);

dfs.getClient().getLocatedFileInfo(testPath.toString(), false);
assertSentTo(2);

dfs.getClient().batchedListPaths(new String[]{"/"}, new byte[0], true);
assertSentTo(2);

// Fake a larger file that needs all 3 data shards
doAnswer((invocation) -> {
List<LocatedBlock> fakeBlocks = new ArrayList<>();
// Return a single location, which is enough for the small file but not for the large file
ExtendedBlock b = new ExtendedBlock("fake-pool", new Block(12345L, 1024 * 3, 0));
DatanodeInfo datanodeInfo = new DatanodeInfo.DatanodeInfoBuilder().build();
LocatedBlock fakeBlock = new LocatedBlock(b, new DatanodeInfo[] {datanodeInfo});
fakeBlocks.add(fakeBlock);
return new LocatedBlocks(1024 * 3, false, fakeBlocks, null, true, null, ecPolicy);
}).when(bmSpy).createLocatedBlocks(Mockito.any(), anyLong(),
anyBoolean(), anyLong(), anyLong(), anyBoolean(), anyBoolean(),
Mockito.any(), Mockito.any());

dfs.open(testPath).close();
assertSentTo(0);

dfs.getClient().listPaths("/", new byte[0], true);
Expand Down