private void doLocalInitiateOffer(Iv2InitiateTaskMessage msg)
    {
        final VoltTrace.TraceEventBatch traceLog = VoltTrace.log(VoltTrace.Category.SPI);
        if (traceLog != null) {
            final String threadName = Thread.currentThread().getName(); // Thread name has to be materialized here
            traceLog.add(() -> VoltTrace.meta("process_name", "name", CoreUtils.getHostnameOrAddress()))
                    .add(() -> VoltTrace.meta("thread_name", "name", threadName))
                    .add(() -> VoltTrace.meta("thread_sort_index", "sort_index", Integer.toString(10000)))
                    .add(() -> VoltTrace.beginAsync("initsp",
                                                    MiscUtils.hsIdPairTxnIdToString(m_mailbox.getHSId(), m_mailbox.getHSId(), msg.getSpHandle(), msg.getClientInterfaceHandle()),
                                                    "ciHandle", msg.getClientInterfaceHandle(),
                                                    "txnId", TxnEgo.txnIdToString(msg.getTxnId()),
                                                    "partition", m_partitionId,
                                                    "read", msg.isReadOnly(),
                                                    "name", msg.getStoredProcedureName(),
                                                    "hsId", CoreUtils.hsIdToString(m_mailbox.getHSId())));
        }

        final String procedureName = msg.getStoredProcedureName();
        final SpProcedureTask task =
            new SpProcedureTask(m_mailbox, procedureName, m_pendingTasks, msg);

        ListenableFuture<Object> durabilityBackpressureFuture =
                m_cl.log(msg, msg.getSpHandle(), null, m_durabilityListener, task);

        if (traceLog != null && durabilityBackpressureFuture != null) {
            traceLog.add(() -> VoltTrace.beginAsync("durability",
                                                    MiscUtils.hsIdTxnIdToString(m_mailbox.getHSId(), msg.getSpHandle()),
                                                    "txnId", TxnEgo.txnIdToString(msg.getTxnId()),
                                                    "partition", Integer.toString(m_partitionId)));
        }

        //Durability future is always null for sync command logging
        //the transaction will be delivered again by the CL for execution once durable
        //Async command logging has to offer the task immediately with a Future for backpressure
        if (m_cl.canOfferTask()) {
            m_pendingTasks.offer(task.setDurabilityBackpressureFuture(durabilityBackpressureFuture));
        }
    }
private void doLocalInitiateOffer ( Iv2InitiateTaskMessage msg ) { final VoltTrace . TraceEventBatch traceLog = VoltTrace . log ( VoltTrace . Category . SPI ) ; if ( traceLog != null ) { final String threadName = Thread . currentThread ( ) . getName ( ) ; traceLog . add ( ( ) -> VoltTrace . meta ( "process_name" , "name" , CoreUtils . getHostnameOrAddress ( ) ) ) . add ( ( ) -> VoltTrace . meta ( "thread_name" , "name" , threadName ) ) . add ( ( ) -> VoltTrace . meta ( "thread_sort_index" , "sort_index" , Integer . toString ( 10000 ) ) ) . add ( ( ) -> VoltTrace . beginAsync ( "initsp" , MiscUtils . hsIdPairTxnIdToString ( m_mailbox . getHSId ( ) , m_mailbox . getHSId ( ) , msg . getSpHandle ( ) , msg . getClientInterfaceHandle ( ) ) , "ciHandle" , msg . getClientInterfaceHandle ( ) , "txnId" , TxnEgo . txnIdToString ( msg . getTxnId ( ) ) , "partition" , m_partitionId , "read" , msg . isReadOnly ( ) , "name" , msg . getStoredProcedureName ( ) , "hsId" , CoreUtils . hsIdToString ( m_mailbox . getHSId ( ) ) ) ) ; } final String procedureName = msg . getStoredProcedureName ( ) ; final SpProcedureTask task = new SpProcedureTask ( m_mailbox , procedureName , m_pendingTasks , msg ) ; ListenableFuture < Object > durabilityBackpressureFuture = m_cl . log ( msg , msg . getSpHandle ( ) , null , m_durabilityListener , task ) ; if ( traceLog != null && durabilityBackpressureFuture != null ) { traceLog . add ( ( ) -> VoltTrace . beginAsync ( "durability" , MiscUtils . hsIdTxnIdToString ( m_mailbox . getHSId ( ) , msg . getSpHandle ( ) ) , "txnId" , TxnEgo . txnIdToString ( msg . getTxnId ( ) ) , "partition" , Integer . toString ( m_partitionId ) ) ) ; } if ( m_cl . canOfferTask ( ) ) { m_pendingTasks . offer ( task . setDurabilityBackpressureFuture ( durabilityBackpressureFuture ) ) ; } }
Do the work necessary to turn the Iv2InitiateTaskMessage into a
TransactionTask which can be queued to the TransactionTaskQueue.
This is reused by both the normal message handling path and the repair
path, and assumes that the caller has dealt with or ensured that the
necessary ID, SpHandles, and replication issues are resolved.