smengcl commented on code in PR #11054:
URL: https://github.com/apache/ozone/pull/11054#discussion_r3822998018
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java:
##########
@@ -536,17 +601,35 @@ public void sendThrottledReplicationCommand(ContainerInfo
containerInfo,
public void sendThrottledReconstructionCommand(ContainerInfo containerInfo,
ReconstructECContainersCommand command)
throws CommandTargetOverloadedException, NotLeaderException {
- List<DatanodeDetails> targets = command.getTargetDatanodes();
- List<Pair<Integer, DatanodeDetails>> targetWithCmds =
- getAvailableDatanodesForReplication(targets);
- if (targetWithCmds.isEmpty()) {
+ if (!tryReserveReconstructionSlot()) {
metrics.incrECReconstructionCmdsDeferredTotal();
- throw new CommandTargetOverloadedException("No target with capacity " +
- "available for reconstruction of " + containerInfo.getContainerID());
+ throw new CommandTargetOverloadedException(
+ "Global reconstruction limit (" + getReconstructionInFlightLimit()
+ + ") reached for container " + containerInfo.getContainerID());
+ }
+ final long cmdId = command.getId();
+ final int fragmentCount = command.getMissingContainerIndexes().size();
+ reconstructionCommandIdToPendingFragmentCount.put(cmdId, fragmentCount);
+ boolean sent = false;
+ try {
+ List<DatanodeDetails> targets = command.getTargetDatanodes();
+ List<Pair<Integer, DatanodeDetails>> targetWithCmds =
+ getAvailableDatanodesForReplication(targets);
+ if (targetWithCmds.isEmpty()) {
+ metrics.incrECReconstructionCmdsDeferredTotal();
+ throw new CommandTargetOverloadedException("No target with capacity " +
+ "available for reconstruction of " +
containerInfo.getContainerID());
+ }
+ DatanodeDetails target = selectAndOptionallyExcludeDatanode(
+ rmConf.getReconstructionCommandWeight(), targetWithCmds);
+ sendDatanodeCommand(command, containerInfo, target);
+ sent = true;
+ } finally {
+ if (!sent) {
+ reconstructionCommandIdToPendingFragmentCount.remove(cmdId);
+ releaseReconstructionSlot();
+ }
}
Review Comment:
**P1 — failover can lose a reconstruction reservation**
`notifyStatusChanged()` clears both the command map and the counter under
`serviceLock`, but this method reserves/registers a command and then calls
`sendDatanodeCommand()` without that lock. A sender can reserve/register, pause
during the leadership transition, and resume after the new leader is ready. It
then sends a command with the new term and schedules ADD ops after the map was
cleared, so those pending fragments are never associated with a tracked
command. The next command can therefore be admitted even though the old
reconstruction is still active, exceeding the configured global cap. Please
serialize the reset with reservation through the send path (or use an
equivalent generation/recheck).
```suggestion
serviceLock.lock();
try {
if (!tryReserveReconstructionSlot()) {
metrics.incrECReconstructionCmdsDeferredTotal();
throw new CommandTargetOverloadedException(
"Global reconstruction limit (" +
getReconstructionInFlightLimit()
+ ") reached for container " +
containerInfo.getContainerID());
}
final long cmdId = command.getId();
final int fragmentCount = command.getMissingContainerIndexes().size();
reconstructionCommandIdToPendingFragmentCount.put(cmdId,
fragmentCount);
boolean sent = false;
try {
List<DatanodeDetails> targets = command.getTargetDatanodes();
List<Pair<Integer, DatanodeDetails>> targetWithCmds =
getAvailableDatanodesForReplication(targets);
if (targetWithCmds.isEmpty()) {
metrics.incrECReconstructionCmdsDeferredTotal();
throw new CommandTargetOverloadedException("No target with
capacity " +
"available for reconstruction of " +
containerInfo.getContainerID());
}
DatanodeDetails target = selectAndOptionallyExcludeDatanode(
rmConf.getReconstructionCommandWeight(), targetWithCmds);
sendDatanodeCommand(command, containerInfo, target);
sent = true;
} finally {
if (!sent) {
reconstructionCommandIdToPendingFragmentCount.remove(cmdId);
releaseReconstructionSlot();
}
}
} finally {
serviceLock.unlock();
}
```
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java:
##########
@@ -1425,6 +1581,21 @@ public void validate() {
"inflight.limit.factor is set to " + inflightReplicationLimitFactor
+ " and must be <= 1");
}
+ if (ecDecommissionReconstructionLoadFactor < 0) {
+ throw new IllegalArgumentException(
+ "decommission.ec.reconstruction.load.factor is set to "
+ + ecDecommissionReconstructionLoadFactor + " and must be >=
0");
+ }
+ if (ecDecommissionReconstructionLoadFactor > 1) {
Review Comment:
**P2 — reject non-finite load factors**
`Double.parseDouble()` accepts `NaN` and infinities, and both range
comparisons are false for `NaN`. The new decommission load factor can therefore
be configured as `NaN`; the existing `inflightReplicationLimitFactor` has the
same gap. Please add `!Double.isFinite(...)` to both validations.
```suggestion
if (!Double.isFinite(ecDecommissionReconstructionLoadFactor)
|| ecDecommissionReconstructionLoadFactor < 0) {
throw new IllegalArgumentException(
"decommission.ec.reconstruction.load.factor is set to "
+ ecDecommissionReconstructionLoadFactor + " and must be >=
0");
}
```
Apply the same finite-value guard to the existing
`inflightReplicationLimitFactor < 0` check above.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
