public void handleInitiateResponseMessage(InitiateResponseMessage message)
    {
        final VoltTrace.TraceEventBatch traceLog = VoltTrace.log(VoltTrace.Category.MPI);
        if (traceLog != null) {
            traceLog.add(() -> VoltTrace.endAsync("initmp", message.getTxnId()));
        }

        DuplicateCounter counter = m_duplicateCounters.get(message.getTxnId());

        // A transaction may be routed back here for EveryPartitionTask via leader migration
        if (counter != null && message.isMisrouted()) {
            tmLog.info("The message on the partition is misrouted. TxnID: " + TxnEgo.txnIdToString(message.getTxnId()));
            Long newLeader = m_leaderMigrationMap.get(message.m_sourceHSId);
            if (newLeader != null) {
                // Update the DuplicateCounter with new replica
                counter.updateReplica(message.m_sourceHSId, newLeader);
                m_leaderMigrationMap.remove(message.m_sourceHSId);

                // Leader migration has updated the leader, send the request to the new leader
                m_mailbox.send(newLeader, counter.getOpenMessage());
            } else {
                // Leader migration not done yet.
                m_mailbox.send(message.m_sourceHSId, counter.getOpenMessage());
            }
            return;
        }

        if (counter != null) {
            int result = counter.offer(message);
            if (result == DuplicateCounter.DONE) {
                m_duplicateCounters.remove(message.getTxnId());
                // Only advance the truncation point on committed transactions that sent fragments to SPIs.
                // See ENG-4211 & ENG-14563
                if (message.shouldCommit() && message.haveSentMpFragment()) {
                    m_repairLogTruncationHandle = m_repairLogAwaitingCommit;
                    m_repairLogAwaitingCommit = message.getTxnId();
                }
                m_outstandingTxns.remove(message.getTxnId());

                m_mailbox.send(counter.m_destinationId, message);
            }
            else if (result == DuplicateCounter.MISMATCH) {
                VoltDB.crashLocalVoltDB("HASH MISMATCH running every-site system procedure.", true, null);
            } else if (result == DuplicateCounter.ABORT) {
                VoltDB.crashLocalVoltDB("PARTIAL ROLLBACK/ABORT running every-site system procedure.", true, null);
            }
            // doing duplicate suppresion: all done.
        }
        else {
            // Only advance the truncation point on committed transactions that sent fragments to SPIs.
            if (message.shouldCommit() && message.haveSentMpFragment()) {
                m_repairLogTruncationHandle = m_repairLogAwaitingCommit;
                m_repairLogAwaitingCommit = message.getTxnId();
            }
            MpTransactionState txn = (MpTransactionState)m_outstandingTxns.remove(message.getTxnId());
            assert(txn != null);
            // the initiatorHSId is the ClientInterface mailbox. Yeah. I know.
            m_mailbox.send(message.getInitiatorHSId(), message);
            // We actually completed this MP transaction.  Create a fake CompleteTransactionMessage
            // to send to our local repair log so that the fate of this transaction is never forgotten
            // even if all the masters somehow die before forwarding Complete on to their replicas.
            CompleteTransactionMessage ctm = new CompleteTransactionMessage(m_mailbox.getHSId(),
                    message.m_sourceHSId, message.getTxnId(), message.isReadOnly(), 0,
                    !message.shouldCommit(), false, false, false, txn.isNPartTxn(),
                    message.m_isFromNonRestartableSysproc, false);
            ctm.setTruncationHandle(m_repairLogTruncationHandle);
            // dump it in the repair log
            // hacky castage
            ((MpInitiatorMailbox)m_mailbox).deliverToRepairLog(ctm);
        }
    }
public void handleInitiateResponseMessage ( InitiateResponseMessage message ) { final VoltTrace . TraceEventBatch traceLog = VoltTrace . log ( VoltTrace . Category . MPI ) ; if ( traceLog != null ) { traceLog . add ( ( ) -> VoltTrace . endAsync ( "initmp" , message . getTxnId ( ) ) ) ; } DuplicateCounter counter = m_duplicateCounters . get ( message . getTxnId ( ) ) ; if ( counter != null && message . isMisrouted ( ) ) { tmLog . info ( "The message on the partition is misrouted. TxnID: " + TxnEgo . txnIdToString ( message . getTxnId ( ) ) ) ; Long newLeader = m_leaderMigrationMap . get ( message . m_sourceHSId ) ; if ( newLeader != null ) { counter . updateReplica ( message . m_sourceHSId , newLeader ) ; m_leaderMigrationMap . remove ( message . m_sourceHSId ) ; m_mailbox . send ( newLeader , counter . getOpenMessage ( ) ) ; } else { m_mailbox . send ( message . m_sourceHSId , counter . getOpenMessage ( ) ) ; } return ; } if ( counter != null ) { int result = counter . offer ( message ) ; if ( result == DuplicateCounter . DONE ) { m_duplicateCounters . remove ( message . getTxnId ( ) ) ; if ( message . shouldCommit ( ) && message . haveSentMpFragment ( ) ) { m_repairLogTruncationHandle = m_repairLogAwaitingCommit ; m_repairLogAwaitingCommit = message . getTxnId ( ) ; } m_outstandingTxns . remove ( message . getTxnId ( ) ) ; m_mailbox . send ( counter . m_destinationId , message ) ; } else if ( result == DuplicateCounter . MISMATCH ) { VoltDB . crashLocalVoltDB ( "HASH MISMATCH running every-site system procedure." , true , null ) ; } else if ( result == DuplicateCounter . ABORT ) { VoltDB . crashLocalVoltDB ( "PARTIAL ROLLBACK/ABORT running every-site system procedure." , true , null ) ; } } else { if ( message . shouldCommit ( ) && message . haveSentMpFragment ( ) ) { m_repairLogTruncationHandle = m_repairLogAwaitingCommit ; m_repairLogAwaitingCommit = message . getTxnId ( ) ; } MpTransactionState txn = ( MpTransactionState ) m_outstandingTxns . remove ( message . getTxnId ( ) ) ; assert ( txn != null ) ; m_mailbox . send ( message . getInitiatorHSId ( ) , message ) ; CompleteTransactionMessage ctm = new CompleteTransactionMessage ( m_mailbox . getHSId ( ) , message . m_sourceHSId , message . getTxnId ( ) , message . isReadOnly ( ) , 0 , ! message . shouldCommit ( ) , false , false , false , txn . isNPartTxn ( ) , message . m_isFromNonRestartableSysproc , false ) ; ctm . setTruncationHandle ( m_repairLogTruncationHandle ) ; ( ( MpInitiatorMailbox ) m_mailbox ) . deliverToRepairLog ( ctm ) ; } }
see all of these messages and control their transmission.