void handleFragmentTaskMessage(FragmentTaskMessage message)
    {
        FragmentTaskMessage msg = message;
        long newSpHandle;
        //The site has been marked as non-leader. The follow-up batches or fragments are processed here
        if (!message.isForReplica() && (m_isLeader || message.isExecutedOnPreviousLeader())) {
            // message processed on leader
            // Quick hack to make progress...we need to copy the FragmentTaskMessage
            // before we start mucking with its state (SPHANDLE).  We need to revisit
            // all the messaging mess at some point.
            msg = new FragmentTaskMessage(message.getInitiatorHSId(),
                    message.getCoordinatorHSId(), message);
            //Not going to use the timestamp from the new Ego because the multi-part timestamp is what should be used
            msg.setTimestamp(message.getTimestamp());
            msg.setExecutedOnPreviousLeader(message.isExecutedOnPreviousLeader());
            if (!message.isReadOnly()) {
                TxnEgo ego = advanceTxnEgo();
                newSpHandle = ego.getTxnId();

                if (m_outstandingTxns.get(msg.getTxnId()) == null) {
                    updateMaxScheduledTransactionSpHandle(newSpHandle);
                }
            } else {
                newSpHandle = getMaxScheduledTxnSpHandle();
            }

            msg.setSpHandle(newSpHandle);
            msg.setLastSpUniqueId(m_uniqueIdGenerator.getLastUniqueId());
            logRepair(msg);
            if (msg.getInitiateTask() != null) {
                msg.getInitiateTask().setSpHandle(newSpHandle);//set the handle
                //Trigger reserialization so the new handle is used
                msg.setStateForDurability(msg.getInitiateTask(), msg.getInvolvedPartitions());
            }

            /*
             * If there a replicas to send it to, forward it!
             * Unless... it's read only AND not a sysproc. Read only sysprocs may expect to be sent
             * everywhere.
             * In that case don't propagate it to avoid a determinism check and extra messaging overhead
             */
            if (IS_KSAFE_CLUSTER && (!message.isReadOnly() || msg.isSysProcTask())) {
                for (long hsId : m_sendToHSIds) {
                    FragmentTaskMessage finalMsg = msg;
                    final VoltTrace.TraceEventBatch traceLog = VoltTrace.log(VoltTrace.Category.SPI);
                    if (traceLog != null) {
                        traceLog.add(() -> VoltTrace.beginAsync("replicatefragment",
                                                                MiscUtils.hsIdPairTxnIdToString(m_mailbox.getHSId(), hsId, finalMsg.getSpHandle(), finalMsg.getTxnId()),
                                                                "txnId", TxnEgo.txnIdToString(finalMsg.getTxnId()),
                                                                "dest", CoreUtils.hsIdToString(hsId)));
                    }
                }

                FragmentTaskMessage replmsg =
                    new FragmentTaskMessage(m_mailbox.getHSId(),
                            m_mailbox.getHSId(), msg);
                replmsg.setForReplica(true);
                replmsg.setTimestamp(msg.getTimestamp());
                // K-safety cluster doesn't always mean partition has replicas,
                // node failure may reduce the number of replicas for each partition.
                if (m_sendToHSIds.length > 0) {
                    m_mailbox.send(m_sendToHSIds,replmsg);
                }
                DuplicateCounter counter;
                /*
                 * Non-determinism should be impossible to happen with MP fragments.
                 * if you see "MP_DETERMINISM_ERROR" as procedure name in the crash logs
                 * something has horribly gone wrong.
                 */
                if (message.getFragmentTaskType() != FragmentTaskMessage.SYS_PROC_PER_SITE) {
                    counter = new DuplicateCounter(
                            msg.getCoordinatorHSId(),
                            msg.getTxnId(),
                            m_replicaHSIds,
                            replmsg);
                }
                else {
                    counter = new SysProcDuplicateCounter(
                            msg.getCoordinatorHSId(),
                            msg.getTxnId(),
                            m_replicaHSIds,
                            replmsg);
                }
                safeAddToDuplicateCounterMap(new DuplicateCounterKey(message.getTxnId(), newSpHandle), counter);
            }
        } else {
            // message processed on replica
            logRepair(msg);
            newSpHandle = msg.getSpHandle();
            setMaxSeenTxnId(newSpHandle);
        }

        Iv2Trace.logFragmentTaskMessage(message, m_mailbox.getHSId(), newSpHandle, false);
        doLocalFragmentOffer(msg);
    }
void handleFragmentTaskMessage ( FragmentTaskMessage message ) { FragmentTaskMessage msg = message ; long newSpHandle ; if ( ! message . isForReplica ( ) && ( m_isLeader || message . isExecutedOnPreviousLeader ( ) ) ) { msg = new FragmentTaskMessage ( message . getInitiatorHSId ( ) , message . getCoordinatorHSId ( ) , message ) ; msg . setTimestamp ( message . getTimestamp ( ) ) ; msg . setExecutedOnPreviousLeader ( message . isExecutedOnPreviousLeader ( ) ) ; if ( ! message . isReadOnly ( ) ) { TxnEgo ego = advanceTxnEgo ( ) ; newSpHandle = ego . getTxnId ( ) ; if ( m_outstandingTxns . get ( msg . getTxnId ( ) ) == null ) { updateMaxScheduledTransactionSpHandle ( newSpHandle ) ; } } else { newSpHandle = getMaxScheduledTxnSpHandle ( ) ; } msg . setSpHandle ( newSpHandle ) ; msg . setLastSpUniqueId ( m_uniqueIdGenerator . getLastUniqueId ( ) ) ; logRepair ( msg ) ; if ( msg . getInitiateTask ( ) != null ) { msg . getInitiateTask ( ) . setSpHandle ( newSpHandle ) ; msg . setStateForDurability ( msg . getInitiateTask ( ) , msg . getInvolvedPartitions ( ) ) ; } if ( IS_KSAFE_CLUSTER && ( ! message . isReadOnly ( ) || msg . isSysProcTask ( ) ) ) { for ( long hsId : m_sendToHSIds ) { FragmentTaskMessage finalMsg = msg ; final VoltTrace . TraceEventBatch traceLog = VoltTrace . log ( VoltTrace . Category . SPI ) ; if ( traceLog != null ) { traceLog . add ( ( ) -> VoltTrace . beginAsync ( "replicatefragment" , MiscUtils . hsIdPairTxnIdToString ( m_mailbox . getHSId ( ) , hsId , finalMsg . getSpHandle ( ) , finalMsg . getTxnId ( ) ) , "txnId" , TxnEgo . txnIdToString ( finalMsg . getTxnId ( ) ) , "dest" , CoreUtils . hsIdToString ( hsId ) ) ) ; } } FragmentTaskMessage replmsg = new FragmentTaskMessage ( m_mailbox . getHSId ( ) , m_mailbox . getHSId ( ) , msg ) ; replmsg . setForReplica ( true ) ; replmsg . setTimestamp ( msg . getTimestamp ( ) ) ; if ( m_sendToHSIds . length > 0 ) { m_mailbox . send ( m_sendToHSIds , replmsg ) ; } DuplicateCounter counter ; if ( message . getFragmentTaskType ( ) != FragmentTaskMessage . SYS_PROC_PER_SITE ) { counter = new DuplicateCounter ( msg . getCoordinatorHSId ( ) , msg . getTxnId ( ) , m_replicaHSIds , replmsg ) ; } else { counter = new SysProcDuplicateCounter ( msg . getCoordinatorHSId ( ) , msg . getTxnId ( ) , m_replicaHSIds , replmsg ) ; } safeAddToDuplicateCounterMap ( new DuplicateCounterKey ( message . getTxnId ( ) , newSpHandle ) , counter ) ; } } else { logRepair ( msg ) ; newSpHandle = msg . getSpHandle ( ) ; setMaxSeenTxnId ( newSpHandle ) ; } Iv2Trace . logFragmentTaskMessage ( message , m_mailbox . getHSId ( ) , newSpHandle , false ) ; doLocalFragmentOffer ( msg ) ; }
doesn't matter, it isn't going to be used for anything.