@Override
    public void afterAborted(TransactionState txnState, boolean txnOperated, String txnStatusChangeReason)
            throws UserException {
        if (!txnOperated) {
            return;
        }

        writeLock();
        try {
            if (isFinalState()) {
                return;
            }
            if (coord != null && !isSyncStreamLoad) {
                coord.cancel(txnStatusChangeReason);
                QeProcessorImpl.INSTANCE.unregisterQuery(loadId);
            }
            for (int i = 0; i < channelNum; i++) {
                this.channels.set(i, State.CANCELLED);
            }
            endTimeMs = System.currentTimeMillis();
            state = State.CANCELLED;
            errorMsg = txnState.getReason();
            gcObject();
            GlobalStateMgr.getCurrentState().getGlobalTransactionMgr().getCallbackFactory().removeCallback(id);
        } finally {
            writeUnlock();
            // sync stream load related query info should unregister here
            QeProcessorImpl.INSTANCE.unregisterQuery(loadId);
        }
    }
@Test
    public void testAfterAborted() throws UserException {
        TransactionState txnState = new TransactionState();
        boolean txnOperated = true;

        TUniqueId labelId = new TUniqueId(2, 3);
        streamLoadTask.setTUniqueId(labelId);
        QeProcessorImpl.INSTANCE.registerQuery(streamLoadTask.getTUniqueId(), coord);
        Assert.assertEquals(1, QeProcessorImpl.INSTANCE.getCoordinatorCount());

        streamLoadTask.afterAborted(txnState, txnOperated, "");
        Assert.assertEquals(0, QeProcessorImpl.INSTANCE.getCoordinatorCount());
    }