@ Override public boolean sequenceForReplay ( VoltMessage message ) { boolean canDeliver = false ; long sequenceWithUniqueId = Long . MIN_VALUE ; boolean commandLog = ( message instanceof TransactionInfoBaseMessage && ( ( ( TransactionInfoBaseMessage ) message ) . isForReplay ( ) ) ) ; boolean sentinel = message instanceof MultiPartitionParticipantMessage ; boolean replay = commandLog || sentinel ; boolean sequenceForReplay = m_isLeader && replay ; if ( replay ) { sequenceWithUniqueId = ( ( TransactionInfoBaseMessage ) message ) . getUniqueId ( ) ; } if ( sequenceForReplay ) { InitiateResponseMessage dupe = m_replaySequencer . dedupe ( sequenceWithUniqueId , ( TransactionInfoBaseMessage ) message ) ; if ( dupe != null ) { // Duplicate initiate task message, send response m_mailbox . send ( dupe . getInitiatorHSId ( ) , dupe ) ; } else if ( ! m_replaySequencer . offer ( sequenceWithUniqueId , ( TransactionInfoBaseMessage ) message ) ) { canDeliver = true ; } else { deliverReadyTxns ( ) ; } // If it's a DR sentinel, send an acknowledgement if ( sentinel && ! commandLog ) { MultiPartitionParticipantMessage mppm = ( MultiPartitionParticipantMessage ) message ; final InitiateResponseMessage response = new InitiateResponseMessage ( mppm ) ; ClientResponseImpl clientResponse = new ClientResponseImpl ( ClientResponseImpl . UNEXPECTED_FAILURE , new VoltTable [ 0 ] , ClientResponseImpl . IGNORED_TRANSACTION ) ; response . setResults ( clientResponse ) ; m_mailbox . send ( response . getInitiatorHSId ( ) , response ) ; } } else { if ( replay ) { // Update last seen and last polled uniqueId for replicas m_replaySequencer . updateLastSeenUniqueId ( sequenceWithUniqueId , ( TransactionInfoBaseMessage ) message ) ; m_replaySequencer . updateLastPolledUniqueId ( sequenceWithUniqueId , ( TransactionInfoBaseMessage ) message ) ; } canDeliver = true ; } return canDeliver ; }
Sequence the message for replay if it s for CL or DR .