HDDS-364. Update open container replica information in SCM during DN register. Contributed by Ajay Kumar.
This commit is contained in:
parent
8563fd67be
commit
a5eba25506
|
@ -505,15 +505,26 @@ public class ContainerMapping implements Mapping {
|
|||
*/
|
||||
@Override
|
||||
public void processContainerReports(DatanodeDetails datanodeDetails,
|
||||
ContainerReportsProto reports)
|
||||
ContainerReportsProto reports, boolean isRegisterCall)
|
||||
throws IOException {
|
||||
List<StorageContainerDatanodeProtocolProtos.ContainerInfo>
|
||||
containerInfos = reports.getReportsList();
|
||||
PendingDeleteStatusList pendingDeleteStatusList =
|
||||
new PendingDeleteStatusList(datanodeDetails);
|
||||
for (StorageContainerDatanodeProtocolProtos.ContainerInfo datanodeState :
|
||||
for (StorageContainerDatanodeProtocolProtos.ContainerInfo contInfo :
|
||||
containerInfos) {
|
||||
byte[] dbKey = Longs.toByteArray(datanodeState.getContainerID());
|
||||
// Update replica info during registration process.
|
||||
if (isRegisterCall) {
|
||||
try {
|
||||
getStateManager().addContainerReplica(ContainerID.
|
||||
valueof(contInfo.getContainerID()), datanodeDetails);
|
||||
} catch (Exception ex) {
|
||||
// Continue to next one after logging the error.
|
||||
LOG.error("Error while adding replica for containerId {}.",
|
||||
contInfo.getContainerID(), ex);
|
||||
}
|
||||
}
|
||||
byte[] dbKey = Longs.toByteArray(contInfo.getContainerID());
|
||||
lock.lock();
|
||||
try {
|
||||
byte[] containerBytes = containerStore.get(dbKey);
|
||||
|
@ -522,12 +533,12 @@ public class ContainerMapping implements Mapping {
|
|||
HddsProtos.SCMContainerInfo.PARSER.parseFrom(containerBytes);
|
||||
|
||||
HddsProtos.SCMContainerInfo newState =
|
||||
reconcileState(datanodeState, knownState, datanodeDetails);
|
||||
reconcileState(contInfo, knownState, datanodeDetails);
|
||||
|
||||
if (knownState.getDeleteTransactionId() > datanodeState
|
||||
if (knownState.getDeleteTransactionId() > contInfo
|
||||
.getDeleteTransactionId()) {
|
||||
pendingDeleteStatusList
|
||||
.addPendingDeleteStatus(datanodeState.getDeleteTransactionId(),
|
||||
.addPendingDeleteStatus(contInfo.getDeleteTransactionId(),
|
||||
knownState.getDeleteTransactionId(),
|
||||
knownState.getContainerID());
|
||||
}
|
||||
|
@ -558,7 +569,7 @@ public class ContainerMapping implements Mapping {
|
|||
LOG.error("Error while processing container report from datanode :" +
|
||||
" {}, for container: {}, reason: container doesn't exist in" +
|
||||
"container database.", datanodeDetails,
|
||||
datanodeState.getContainerID());
|
||||
contInfo.getContainerID());
|
||||
}
|
||||
} finally {
|
||||
lock.unlock();
|
||||
|
|
|
@ -84,7 +84,8 @@ public class ContainerReportHandler implements
|
|||
try {
|
||||
|
||||
//update state in container db and trigger close container events
|
||||
containerMapping.processContainerReports(datanodeOrigin, containerReport);
|
||||
containerMapping
|
||||
.processContainerReports(datanodeOrigin, containerReport, false);
|
||||
|
||||
Set<ContainerID> containerIds = containerReport.getReportsList().stream()
|
||||
.map(containerProto -> containerProto.getContainerID())
|
||||
|
|
|
@ -115,7 +115,7 @@ public interface Mapping extends Closeable {
|
|||
* @param reports Container report
|
||||
*/
|
||||
void processContainerReports(DatanodeDetails datanodeDetails,
|
||||
ContainerReportsProto reports)
|
||||
ContainerReportsProto reports, boolean isRegisterCall)
|
||||
throws IOException;
|
||||
|
||||
/**
|
||||
|
|
|
@ -196,7 +196,7 @@ public class SCMDatanodeProtocolServer implements
|
|||
if (registeredCommand.getError()
|
||||
== SCMRegisteredResponseProto.ErrorCode.success) {
|
||||
scm.getScmContainerManager().processContainerReports(datanodeDetails,
|
||||
containerReportsProto);
|
||||
containerReportsProto, true);
|
||||
}
|
||||
return getRegisteredResponse(registeredCommand);
|
||||
}
|
||||
|
|
|
@ -242,7 +242,7 @@ public class TestContainerMapping {
|
|||
}
|
||||
|
||||
@Test
|
||||
public void testFullContainerReport() throws IOException {
|
||||
public void testFullContainerReport() throws Exception {
|
||||
ContainerInfo info = createContainer();
|
||||
DatanodeDetails datanodeDetails = TestUtils.randomDatanodeDetails();
|
||||
List<StorageContainerDatanodeProtocolProtos.ContainerInfo> reports =
|
||||
|
@ -266,13 +266,26 @@ public class TestContainerMapping {
|
|||
.newBuilder();
|
||||
crBuilder.addAllReports(reports);
|
||||
|
||||
mapping.processContainerReports(datanodeDetails, crBuilder.build());
|
||||
mapping.processContainerReports(datanodeDetails, crBuilder.build(), false);
|
||||
|
||||
ContainerInfo updatedContainer =
|
||||
mapping.getContainer(info.getContainerID());
|
||||
Assert.assertEquals(100000000L,
|
||||
updatedContainer.getNumberOfKeys());
|
||||
Assert.assertEquals(2000000000L, updatedContainer.getUsedBytes());
|
||||
|
||||
for (StorageContainerDatanodeProtocolProtos.ContainerInfo c : reports) {
|
||||
LambdaTestUtils.intercept(SCMException.class, "No entry "
|
||||
+ "exist for containerId:", () -> mapping.getStateManager()
|
||||
.getContainerReplicas(ContainerID.valueof(c.getContainerID())));
|
||||
}
|
||||
|
||||
mapping.processContainerReports(TestUtils.randomDatanodeDetails(),
|
||||
crBuilder.build(), true);
|
||||
for (StorageContainerDatanodeProtocolProtos.ContainerInfo c : reports) {
|
||||
Assert.assertTrue(mapping.getStateManager().getContainerReplicas(
|
||||
ContainerID.valueof(c.getContainerID())).size() > 0);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
|
@ -301,7 +314,7 @@ public class TestContainerMapping {
|
|||
ContainerReportsProto.newBuilder();
|
||||
crBuilder.addAllReports(reports);
|
||||
|
||||
mapping.processContainerReports(datanodeDetails, crBuilder.build());
|
||||
mapping.processContainerReports(datanodeDetails, crBuilder.build(), false);
|
||||
|
||||
ContainerInfo updatedContainer =
|
||||
mapping.getContainer(info.getContainerID());
|
||||
|
|
|
@ -223,6 +223,6 @@ public class TestContainerCloser {
|
|||
.setDeleteTransactionId(0);
|
||||
reports.addReports(ciBuilder);
|
||||
mapping.processContainerReports(TestUtils.randomDatanodeDetails(),
|
||||
reports.build());
|
||||
reports.build(), false);
|
||||
}
|
||||
}
|
|
@ -187,7 +187,8 @@ public class TestBlockDeletion {
|
|||
|
||||
logCapturer.clearOutput();
|
||||
scm.getScmContainerManager().processContainerReports(
|
||||
cluster.getHddsDatanodes().get(0).getDatanodeDetails(), dummyReport);
|
||||
cluster.getHddsDatanodes().get(0).getDatanodeDetails(), dummyReport,
|
||||
false);
|
||||
// wait for event to be handled by event handler
|
||||
Thread.sleep(1000);
|
||||
String output = logCapturer.getOutput();
|
||||
|
|
Loading…
Reference in New Issue