void send(TXLockId lockId) {
    if (isEmpty()) {
      if (logger.isDebugEnabled()) {
        logger.debug("empty transaction - nothing to distribute");
      }
      return;
    }
    Assert.assertTrue(txState != null, "Send must have transaction state.");
    this.lockId = (TXLockIdImpl) lockId;
    updateLockMembers();

    // Map of RegionCommitList keys to Sets of receivers
    IdentityHashMap<RegionCommitList, Set<InternalDistributedMember>> distMap =
        new IdentityHashMap<>();
    HashSet<InternalDistributedMember> ackReceivers = null;
    for (final Map.Entry<InternalDistributedMember, RegionCommitList> entry : msgMap.entrySet()) {
      final RegionCommitList rcl = entry.getValue();
      if (rcl.getNeedsAck()) {
        if (ackReceivers == null) {
          ackReceivers = new HashSet<>();
        }
        ackReceivers.add(entry.getKey());
      }
      final Set<InternalDistributedMember> receivers =
          distMap.computeIfAbsent(rcl, k -> new HashSet<>());
      receivers.add(entry.getKey());
    }

    CommitReplyProcessor processor = null;
    {
      if (ackReceivers != null) {
        processor = new CommitReplyProcessor(dm, ackReceivers, msgMap);
        if (ackReceivers.size() > 1) {
          farSiders = ackReceivers;
        }
        processor.enableSevereAlertProcessing();
      }
      {
        for (final Map.Entry<RegionCommitList, Set<InternalDistributedMember>> me : distMap
            .entrySet()) {
          RegionCommitList rcl = me.getKey();
          Set<InternalDistributedMember> recipients = me.getValue();
          // now remove from the recipients any recipients that the dm no
          // longer knows about
          recipients.retainAll(dm.getDistributionManagerIds());
          if (!recipients.isEmpty()) {
            if (txState.internalDuringIndividualSend != null) {
              // Run in test mode, splitting out individual recipients,
              // so we can control who gets what
              for (final InternalDistributedMember recipient : recipients) {
                txState.internalDuringIndividualSend.run();
                setRecipientsSendData(Collections.singleton(recipient), processor, rcl);
              }
            } else {
              // Run in normal mode sending to multiple recipients in one shot
              setRecipientsSendData(recipients, processor, rcl);
            }
          }
        }
      }
    }
    if (txState.internalAfterIndividualSend != null) {
      txState.internalAfterIndividualSend.run();
    }

    if (processor != null) {
      // Send the CommitProcessMessage
      final CommitProcessMessage cpMsg;
      if (this.lockId != null) {
        cpMsg = new CommitProcessForLockIdMessage(this.lockId);
      } else {
        cpMsg = new CommitProcessForTXIdMessage(txIdent);
      }
      if (txState.internalDuringIndividualCommitProcess != null) {
        // Run in test mode
        for (final InternalDistributedMember ackReceiver : ackReceivers) {
          txState.internalDuringIndividualCommitProcess.run();
          cpMsg.setRecipients(Collections.singleton(ackReceiver));
          dm.putOutgoing(cpMsg);
          cpMsg.resetRecipients();
        }
      } else {
        // Run in normal mode
        cpMsg.setRecipients(ackReceivers);
        dm.putOutgoing(cpMsg);
      }

      if (txState.internalAfterIndividualCommitProcess != null) {
        // Testing callback
        txState.internalAfterIndividualCommitProcess.run();
      }

      // for() loop removed for bug 36983 - you can't loop on waitForReplies()
      dm.getCancelCriterion().checkCancelInProgress(null);
      processor.waitForCommitCompletion();
      dm.getStats().incCommitWaits();
    }
    if (hasReliableRegions) {
      checkDistributionReliability(distMap, processor);
    }
  }
record CacheDistributionAdvisor.startOperation versions for later cleanup
geode-core/src/main/java/org/apache/geode/internal/cache/TXCommitMessage.java