@Override
    public void replayOnAborted(TransactionState txnState) {
        writeLock();
        try {
            replayTxnAttachment(txnState);
            failMsg = new FailMsg(FailMsg.CancelType.LOAD_RUN_FAIL, txnState.getReason());
            finishTimestamp = txnState.getFinishTime();
            state = JobState.CANCELLED;
            if (retryTime <= 0 || !failMsg.getMsg().contains("timeout") || !isTimeout()) {
                GlobalStateMgr.getCurrentState().getGlobalTransactionMgr().getCallbackFactory().removeCallback(id);
                return;
            }
            retryTime--;
        } finally {
            writeUnlock();
        }
    }
@Test
    public void testReplayOnAbortedAfterFailure(@Injectable TransactionState txnState,
                                                @Injectable LoadJobFinalOperation attachment,
                                                @Injectable FailMsg failMsg) {
        BrokerLoadJob brokerLoadJob = new BrokerLoadJob();
        brokerLoadJob.setId(1);
        GlobalTransactionMgr globalTxnMgr = GlobalStateMgr.getCurrentState().getGlobalTransactionMgr();
        globalTxnMgr.getCallbackFactory().addCallback(brokerLoadJob);

        // 1. The job will be keep when the failure is timeout
        new Expectations() {
            {
                txnState.getTxnCommitAttachment();
                minTimes = 0;
                result = attachment;
                txnState.getReason();
                minTimes = 0;
                result = "load timeout";
            }
        };

        brokerLoadJob.replayOnAborted(txnState);
        TxnStateChangeCallback callback = globalTxnMgr.getCallbackFactory().getCallback(1);
        Assert.assertNotNull(callback);

        // 2. The job will be discard when failure isn't timeout
        new Expectations() {
            {
                txnState.getTxnCommitAttachment();
                minTimes = 0;
                result = attachment;
                txnState.getReason();
                minTimes = 0;
                result = "load_run_fail";
            }
        };
        brokerLoadJob.replayOnAborted(txnState);
        callback = globalTxnMgr.getCallbackFactory().getCallback(1);
        Assert.assertNull(callback);
    }