sodonnel commented on code in PR #3990:
URL: https://github.com/apache/ozone/pull/3990#discussion_r1029884882
##########
hadoop-hdds/server-scm/src/main/java/org/apache/hadoop/hdds/scm/container/replication/ReplicationManager.java:
##########
@@ -401,22 +403,54 @@ public void sendCloseContainerEvent(ContainerID
containerID) {
*/
public void sendDeleteCommand(final ContainerInfo container, int
replicaIndex,
final DatanodeDetails datanode) throws NotLeaderException {
- LOG.info("Sending delete container command for container {}" +
- " to datanode {}", container.containerID(), datanode);
-
final DeleteContainerCommand deleteCommand =
new DeleteContainerCommand(container.containerID(), false);
- deleteCommand.setTerm(getScmTerm());
-
- final CommandForDatanode<DeleteContainerCommandProto> datanodeCommand =
- new CommandForDatanode<>(datanode.getUuid(), deleteCommand);
+ deleteCommand.setReplicaIndex(replicaIndex);
+ sendDatanodeCommand(deleteCommand, container, datanode);
+ }
+
+ public void sendDatanodeCommand(SCMCommand<?> command,
+ ContainerInfo containerInfo, DatanodeDetails target)
+ throws NotLeaderException {
+ LOG.info("Sending command of type {} for container {} to {}",
+ command.getType(), containerInfo, target);
+ command.setTerm(getScmTerm());
+ final CommandForDatanode<?> datanodeCommand =
+ new CommandForDatanode<>(target.getUuid(), command);
eventPublisher.fireEvent(SCMEvents.DATANODE_COMMAND, datanodeCommand);
- containerReplicaPendingOps.scheduleDeleteReplica(container.containerID(),
- datanode, replicaIndex);
-
- synchronized (this) {
- metrics.incrNumDeletionCmdsSent();
- metrics.incrNumDeletionBytesTotal(container.getUsedBytes());
+ adjustPendingOpsAndMetrics(containerInfo, command, target);
+ }
+
+ private void adjustPendingOpsAndMetrics(ContainerInfo containerInfo,
+ SCMCommand<?> cmd, DatanodeDetails targetDatanode) {
+ if (cmd.getType() == Type.deleteContainerCommand) {
+ DeleteContainerCommand rcc = (DeleteContainerCommand) cmd;
+ containerReplicaPendingOps.scheduleDeleteReplica(
+ containerInfo.containerID(), targetDatanode, rcc.getReplicaIndex());
+ if (rcc.getReplicaIndex() > 0) {
+ getMetrics().incrEcDeletionCmdsSentTotal();
+ } else if (rcc.getReplicaIndex() == 0) {
+ getMetrics().incrNumDeletionCmdsSent();
+ getMetrics().incrNumDeletionBytesTotal(containerInfo.getUsedBytes());
+ }
+ } else if (cmd.getType() == Type.reconstructECContainersCommand) {
+ ReconstructECContainersCommand rcc = (ReconstructECContainersCommand)
cmd;
+ List<DatanodeDetails> targets = rcc.getTargetDatanodes();
+ byte[] targetIndexes = rcc.getMissingContainerIndexes();
+ for (int i = 0; i < targetIndexes.length; i++) {
+ containerReplicaPendingOps.scheduleAddReplica(
+ containerInfo.containerID(), targets.get(i), targetIndexes[i]);
+ }
+ getMetrics().incrEcReconstructionCmdsSentTotal();
+ } else if (cmd.getType() == Type.replicateContainerCommand) {
+ ReplicateContainerCommand rcc = (ReplicateContainerCommand) cmd;
+ containerReplicaPendingOps.scheduleAddReplica(
+ containerInfo.containerID(), targetDatanode, rcc.getReplicaIndex());
+ if (rcc.getReplicaIndex() > 0) {
+ getMetrics().incrEcReplicationCmdsSentTotal();
+ } else if (rcc.getReplicaIndex() == 0) {
+ getMetrics().incrNumReplicationCmdsSent();
+ }
Review Comment:
yes, that is why I removed that part. Anything which does need ops or
metrics updated just passes through.
--
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]
