smengcl commented on code in PR #11307:
URL: https://github.com/apache/ozone/pull/11307#discussion_r4319110499
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/SCMDeletedBlockTransactionStatusManager.java:
##########
@@ -628,6 +664,63 @@ public void
commitTransactions(List<DeleteBlockTransactionResult> transactionRes
}
}
+ private boolean hasPendingAddReplica(ContainerID containerId) {
+ return
containerReplicaPendingOpsSupplier.get().getPendingOps(containerId).stream()
+ .anyMatch(op -> op.getOpType() ==
ContainerReplicaOp.PendingOpType.ADD);
+ }
+
+ /**
+ * Whether a transaction whose purge {@link #commitTransactions} deferred
can be purged without another ACK:
+ * no replica ADD is in flight and every current replica has already
committed it. A deferral caused only by
+ * an in-flight copy that later fails or expires is otherwise never retried,
since no replica is left to ACK
+ * a resend and the entry would stay in {@link #transactionToDNsCommitMap}
indefinitely.
+ */
+ boolean isPurgeableWithoutAck(long txID, ContainerID containerId) {
+ final Set<DatanodeID> committedDns = transactionToDNsCommitMap.get(txID);
+ if (committedDns == null || committedDns.isEmpty() ||
hasPendingAddReplica(containerId)) {
+ return false;
+ }
+ try {
+ final Set<ContainerReplica> replicas =
containerManager.getContainerReplicas(containerId);
+ return !replicas.isEmpty()
+ && min(replicas.size(), committedDns.size())
+ >=
containerManager.getContainer(containerId).getReplicationConfig().getRequiredNodes()
+ && replicas.stream().map(r ->
r.getDatanodeDetails().getID()).allMatch(committedDns::contains);
+ } catch (ContainerNotFoundException e) {
+ return false;
+ }
+ }
+
+ /**
+ * Checks that every replica in the pre-purge snapshot has committed the
deletion. A missing snapshot means the
+ * container has an in-flight replica ADD or could not be looked up, so the
transaction is kept for resend.
+ *
+ * <p>Not covered: a copy exported before the deletion was recorded whose
pending ADD expired before its replica
Review Comment:
Optional, not blocking. SCM expires the ADD after
`hdds.scm.replication.event.timeout` (12m), the datanode only has to start
within 12m minus the 6m offset, and the import itself is not bounded (5 GB at
10 MB/s takes over 8 minutes).
Purging only when the container is HEALTHY, as `getTransactions` already
requires via `checkInadequateReplica`, would cover copies made for under
replication, since the container stays under replicated until a copy registers.
HDDS-17152 would then only need the case of a late copy registering after a
replacement copy.
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/SCMDeletedBlockTransactionStatusManager.java:
##########
@@ -628,6 +664,63 @@ public void
commitTransactions(List<DeleteBlockTransactionResult> transactionRes
}
}
+ private boolean hasPendingAddReplica(ContainerID containerId) {
+ return
containerReplicaPendingOpsSupplier.get().getPendingOps(containerId).stream()
+ .anyMatch(op -> op.getOpType() ==
ContainerReplicaOp.PendingOpType.ADD);
+ }
+
+ /**
+ * Whether a transaction whose purge {@link #commitTransactions} deferred
can be purged without another ACK:
+ * no replica ADD is in flight and every current replica has already
committed it. A deferral caused only by
+ * an in-flight copy that later fails or expires is otherwise never retried,
since no replica is left to ACK
+ * a resend and the entry would stay in {@link #transactionToDNsCommitMap}
indefinitely.
+ */
+ boolean isPurgeableWithoutAck(long txID, ContainerID containerId) {
+ final Set<DatanodeID> committedDns = transactionToDNsCommitMap.get(txID);
+ if (committedDns == null || committedDns.isEmpty() ||
hasPendingAddReplica(containerId)) {
+ return false;
+ }
+ try {
+ final Set<ContainerReplica> replicas =
containerManager.getContainerReplicas(containerId);
+ return !replicas.isEmpty()
+ && min(replicas.size(), committedDns.size())
+ >=
containerManager.getContainer(containerId).getReplicationConfig().getRequiredNodes()
+ && replicas.stream().map(r ->
r.getDatanodeDetails().getID()).allMatch(committedDns::contains);
Review Comment:
Same check as line 675 to 681. Can both use one helper?
`!replicas.isEmpty()` is redundant since `min(0, n) >= requiredNodes` is false.
Pls also note why replicas are re-read here instead of reusing the set from
`getTransactions`: that set was read before the pending ADD check.
```suggestion
// Read after the pending ADD check: an earlier read can miss a
replica whose ADD completed in between.
final Set<ContainerReplica> replicas =
containerManager.getContainerReplicas(containerId);
return
allReplicasCommitted(containerManager.getContainer(containerId), replicas,
committedDns);
```
```diff
+ private static boolean allReplicasCommitted(ContainerInfo container,
Set<ContainerReplica> replicas,
+ Set<DatanodeID> committedDns) {
+ return min(replicas.size(), committedDns.size()) >=
container.getReplicationConfig().getRequiredNodes()
+ && replicas.stream().map(r ->
r.getDatanodeDetails().getID()).allMatch(committedDns::contains);
+ }
```
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/block/SCMDeletedBlockTransactionStatusManager.java:
##########
@@ -619,6 +624,37 @@ public void
commitTransactions(List<DeleteBlockTransactionResult> transactionRes
transactionResult.getTxID(), e);
}
}
+ // A container copy made by ReplicationManager misses this deletion only
if it was exported before the source
+ // recorded the deletion: the datanode serializes export and recording
under the container lock, and ACKs only
+ // after recording. Such a copy was therefore scheduled before the ACKs
above, so its replica ADD is either still
+ // pending here or already completed. ContainerStateManager registers a
replica and clears its pending ADD under
+ // the same container lock, so the pending ADD is checked before the
replica set is read: an ADD that is no
+ // longer pending is already in that replica set. A copy scheduled after
this check is exported after the
+ // deletion was recorded and deletes the blocks without a resend.
+ final Map<ContainerID, List<DatanodeID>> replicaSnapshots = new
HashMap<>();
+ for (ContainerID containerId : new
HashSet<>(purgeCandidateContainers.values())) {
+ if (hasPendingAddReplica(containerId)) {
+ LOG.info("Deferring purge of transactions for container {}: a replica
is being added to it.", containerId);
+ continue;
+ }
+ try {
+ replicaSnapshots.put(containerId,
containerManager.getContainerReplicas(containerId).stream()
+ .map(ContainerReplica::getDatanodeDetails)
+ .map(DatanodeDetails::getID)
+ .collect(Collectors.toList()));
+ } catch (ContainerNotFoundException e) {
+ LOG.warn("Could not re-validate replicas for container {} before
purge; deferring its transactions.",
+ containerId, e);
+ }
+ }
+ txIDsToBeDeleted.removeIf(txID -> !isSafeToPurge(txID,
purgeCandidateContainers.get(txID), replicaSnapshots));
+ for (Long txID : txIDsToBeDeleted) {
+ transactionToDNsCommitMap.remove(txID);
+ transactionToRetryCountMap.remove(txID);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Purging txId: {} from block deletion log", txID);
+ }
+ }
Review Comment:
This second pass isn't needed if the pending ADD check moves before the
in-loop `getContainerReplicas()`. An ADD that is not pending at the check has
either registered (so the read sees it) or was scheduled after it, when its
source had already recorded the deletion. That drops
`purgeCandidateContainers`, `replicaSnapshots`, `isSafeToPurge` and this block,
plus `testCommitTransactionsDefersPurgeForReplicaAddedDuringCommit` and
`testCommitTransactionsReadsEachContainerOnceAndDefersConsistently`.
```diff
dnsWithCommittedTxn.add(dnId);
+ if (hasPendingAddReplica(containerId)) {
+ continue;
+ }
final ContainerInfo container =
@@
if (dnsWithCommittedTxn.containsAll(containerDns)) {
- purgeCandidateContainers.put(txID, containerId);
+ transactionToDNsCommitMap.remove(txID);
+ transactionToRetryCountMap.remove(txID);
txIDsToBeDeleted.add(txID);
```
##########
hadoop-hdds/server-scm/src/test/java/org/apache/hadoop/hdds/scm/block/TestDeletedBlockLog.java:
##########
@@ -412,6 +419,199 @@ public void testCommitTransactions() throws Exception {
assertEquals(0, blocks.size());
}
+ // The tests below exercise SCM's purge decision in
SCMDeletedBlockTransactionStatusManager#commitTransactions.
+ // TestKeyValueContainer#testExportCarriesOnlyDeletionsRecordedBeforeExport
covers the datanode side: a copy
+ // exported before the source recorded the deletion keeps the blocks with no
pending-delete record, while a copy
+ // exported later carries it. These tests verify that SCM keeps the
transaction for resend whenever it cannot rule
+ // out a replica of the first kind.
+ @Test
+ public void testCommitTransactionsDefersPurgeForReplicaAddedDuringCommit()
+ throws Exception {
+ deletedBlockLog.setScmCommandTimeoutMs(Long.MAX_VALUE);
+ addTransactions(generateData(1), true);
+ mockContainerHealthResult(true);
+ List<DeletedBlocksTransaction> blocks =
+ getTransactions(BLOCKS_PER_TXN * THREE);
+ assertEquals(THREE, blocks.size());
+ // All THREE entries above are the same logical transaction, one per
replica DN;
+ // commit it one DN at a time to control exactly how many times
+ // containerManager.getContainerReplicas() is invoked per commit call.
+ List<DeletedBlocksTransaction> singleTxn =
+ Collections.singletonList(blocks.get(0));
+
+ final ContainerID cid =
ContainerID.valueOf(blocks.get(0).getContainerID());
+ final Set<ContainerReplica> originalReplicas = new
HashSet<>(replicas.get(cid));
+
+ // Simulate ReplicationManager landing a new, never-acked replica of the
container
+ // in the window between the ack-set check and the durable purge: the
first three
+ // getContainerReplicas() calls (the in-loop decision for each of the
three commits
+ // below) see the original replica set, while every call after that (the
pre-purge
+ // recheck) sees a fourth replica that never received or acknowledged the
deletion.
+ ContainerReplica newReplica = ContainerReplica.newBuilder()
+ .setContainerID(cid)
+ .setContainerState(ContainerReplicaProto.State.CLOSED)
+
.setDatanodeDetails(DatanodeDetails.newBuilder().setUuid(UUID.randomUUID()).build())
+ .build();
+ Set<ContainerReplica> grownReplicas = new HashSet<>(originalReplicas);
+ grownReplicas.add(newReplica);
+ when(containerManager.getContainerReplicas(cid)).thenReturn(
+ originalReplicas, originalReplicas, originalReplicas, grownReplicas);
+
+ commitTransactions(singleTxn, dnList.get(0));
+ commitTransactions(singleTxn, dnList.get(1));
+ commitTransactions(singleTxn, dnList.get(2));
+
+ // The transaction must remain outstanding: the new replica has not
acknowledged the
+ // delete, so it may be a copy exported before the source recorded the
deletion.
+ assertEquals(1, deletedBlockLog.getNumOfValidTransactions(),
+ "transaction was purged although a never-acked replica exists");
+ assertEquals(1, deletedBlockLog.getTransactionToDNsCommitMapSize(),
+ "commit tracking was dropped although the transaction was not purged");
+ }
+
+ @Test
+ public void
testCommitTransactionsReadsEachContainerOnceAndDefersConsistently()
+ throws Exception {
+ deletedBlockLog.setScmCommandTimeoutMs(Long.MAX_VALUE);
+ mockContainerHealthResult(true);
+
+ // Two independent delete-block transactions against the SAME container (a
container
+ // commonly accumulates several separate pending deletions over time),
each already
+ // acknowledged by two of its three replicas before the decisive final ack
below.
+ Map<Long, List<DeletedBlock>> firstTxBlocks = generateData(1);
+ long containerID = firstTxBlocks.keySet().iterator().next();
+ List<DeletedBlock> secondTxBlocks = firstTxBlocks.get(containerID).stream()
+ .map(b -> new DeletedBlock(new BlockID(containerID,
b.getBlockID().getLocalID() + BLOCKS_PER_TXN),
+ b.getSize(), b.getReplicatedSize(), b.getSizePerReplica()))
+ .collect(Collectors.toList());
+ addTransactions(firstTxBlocks, true);
+ addTransactions(Collections.singletonMap(containerID, secondTxBlocks),
true);
+
+ List<DeletedBlocksTransaction> blocks = getTransactions(2 * BLOCKS_PER_TXN
* THREE);
+ assertEquals(2 * THREE, blocks.size());
+ Collection<DeletedBlocksTransaction> bothTxns = blocks.stream()
+ .collect(Collectors.toMap(DeletedBlocksTransaction::getTxID, t -> t,
(a, b) -> a))
+ .values();
+ assertEquals(2, bothTxns.size());
+
+ final ContainerID cid = ContainerID.valueOf(containerID);
+ final Set<ContainerReplica> originalReplicas = new
HashSet<>(replicas.get(cid));
+
+ commitTransactions(bothTxns, dnList.get(0));
+ commitTransactions(bothTxns, dnList.get(1));
+
+ // Simulate ReplicationManager landing a new, never-acked replica exactly
as the
+ // decisive final ack arrives: the ack-set check for each of the two txIDs
below sees
+ // the original replica set, while the pre-purge snapshot sees a fourth
replica that
+ // never received or acknowledged either deletion.
+ ContainerReplica newReplica = ContainerReplica.newBuilder()
+ .setContainerID(cid)
+ .setContainerState(ContainerReplicaProto.State.CLOSED)
+
.setDatanodeDetails(DatanodeDetails.newBuilder().setUuid(UUID.randomUUID()).build())
+ .build();
+ Set<ContainerReplica> grownReplicas = new HashSet<>(originalReplicas);
+ grownReplicas.add(newReplica);
+ when(containerManager.getContainerReplicas(cid)).thenReturn(
+ originalReplicas, originalReplicas, grownReplicas);
+ clearInvocations(containerManager);
+
+ commitTransactions(bothTxns, dnList.get(2));
+
+ // Two ack-set checks (one per txID) plus exactly one deduplicated
pre-purge snapshot
+ // shared by both txIDs of this container: three calls total, not four.
+ verify(containerManager, times(3)).getContainerReplicas(cid);
Review Comment:
nit: drop this `verify`. It pins the call count, which the `thenReturn(...)`
stubbing above already depends on
--
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]
