@Override
    public void onFinished(final MessageRecord messageRecord) {
        try {
            Preconditions.checkState(messageRecord != null, "Fetched msg is null.");
            CacheMessageRecord cacheRecord = new CacheMessageRecord(messageRecord, client,
                    CommonPropertiesHolder.getAckPolicy());
            for (InLongMessage inLongMessage : messageRecord.getMsgs()) {
                SortTaskStatusRepository.acquirePutChannel(sortTaskName);
                final ProfileEvent profileEvent = new ProfileEvent(inLongMessage, cacheRecord);
                channelProcessor.processEvent(profileEvent);
                context.reportToMetric(profileEvent, sortTaskName, "-", SortSdkSourceContext.FetchResult.SUCCESS);
            }
        } catch (NullPointerException npe) {
            LOG.error("Got a null pointer exception for sort task " + sortTaskName, npe);
            context.reportToMetric(null, sortTaskName, "-", SortSdkSourceContext.FetchResult.FAILURE);
        } catch (Exception e) {
            LOG.error("Got exception of sort task " + sortTaskName, e);
            context.reportToMetric(null, sortTaskName, "-", SortSdkSourceContext.FetchResult.FAILURE);
        }
    }
The callback function that SortSDK invoke when fetch messages.

<p> In order to ACK the fetched msg, {@link FetchCallback} has to hold the {@link SortClient} which results in
<b>Cycle Reference</b>. The {@link SortClient} should deliver one object to ACK after invoke the callback method
{@link ReadCallback#onFinished(MessageRecord)}. </p>

@param messageRecord message
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/FetchCallback.java