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

        // sync stream load collect profile, here we collect profile only when be has reported
        if (isSyncStreamLoad() && coord != null && coord.isProfileAlreadyReported()) {
            collectProfile();
        }

        writeLock();
        try {
            for (int i = 0; i < channelNum; i++) {
                this.channels.set(i, State.COMMITED);
            }
            this.state = State.COMMITED;
            isCommitting = false;
            endTimeMs = System.currentTimeMillis();
        } finally {
            writeUnlock();
            // sync stream load related query info should unregister here
            QeProcessorImpl.INSTANCE.unregisterQuery(loadId);
        }
    }
@Test
    public void testAfterCommitted() throws UserException {
        streamLoadTask.setCoordinator(coord);
        new Expectations() {
            {
                coord.isProfileAlreadyReported();
                result = false;
            }
        };
        TUniqueId labelId = new TUniqueId(2, 3);
        streamLoadTask.setTUniqueId(labelId);
        QeProcessorImpl.INSTANCE.registerQuery(streamLoadTask.getTUniqueId(), coord);
        Assert.assertEquals(1, QeProcessorImpl.INSTANCE.getCoordinatorCount());

        TransactionState txnState = new TransactionState();
        boolean txnOperated = true;
        streamLoadTask.afterCommitted(txnState, txnOperated);
        Assert.assertEquals(0, QeProcessorImpl.INSTANCE.getCoordinatorCount());
    }