private SortClient newClient(final String sortTaskName) {
        LOG.info("Start to new sort client for task: {}", sortTaskName);
        try {
            final SortClientConfig clientConfig = new SortClientConfig(sortTaskName, this.sortClusterName,
                    new DefaultTopicChangeListener(),
                    SortSdkSource.defaultStrategy, InetAddress.getLocalHost().getHostAddress());
            final FetchCallback callback = FetchCallback.Factory.create(sortTaskName, getChannelProcessor(), context);
            clientConfig.setCallback(callback);
            Map<String, String> sortSdkParams = this.getSortClientConfigParameters();
            clientConfig.setParameters(sortSdkParams);

            // create SortClient
            String configType = CommonPropertiesHolder
                    .getString(SortSourceConfigType.KEY_TYPE, SortSourceConfigType.MANAGER.name());
            SortClient client = null;
            if (SortClusterConfigType.FILE.name().equalsIgnoreCase(configType)) {
                LOG.info("Create sort sdk client in file way:{}", configType);
                ClassResourceQueryConsumeConfig queryConfig = new ClassResourceQueryConsumeConfig();
                client = SortClientFactory.createSortClient(clientConfig, queryConfig);
            } else if (SortClusterConfigType.MANAGER.name().equalsIgnoreCase(configType)) {
                LOG.info("Create sort sdk client in manager way:{}", configType);
                clientConfig.setManagerApiUrl(ManagerUrlHandler.getSortSourceConfigUrl());
                client = SortClientFactory.createSortClient(clientConfig);
            } else {
                LOG.info("Create sort sdk client in custom way:{}", configType);
                // user-defined
                Class<?> loaderClass = ClassUtils.getClass(configType);
                Object loaderObject = loaderClass.getDeclaredConstructor().newInstance();
                if (loaderObject instanceof Configurable) {
                    ((Configurable) loaderObject).configure(new Context(CommonPropertiesHolder.get()));
                }
                if (!(loaderObject instanceof QueryConsumeConfig)) {
                    LOG.error("Got exception when create QueryConsumeConfig instance,config key:{},config class:{}",
                            SortSourceConfigType.KEY_TYPE, configType);
                    return null;
                }
                // if it specifies the type of QueryConsumeConfig.
                client = SortClientFactory.createSortClient(clientConfig, (QueryConsumeConfig) loaderObject);
            }

            // init
            client.init();
            // temporary use to ACK fetched msg.
            callback.setClient(client);
            return client;
        } catch (UnknownHostException ex) {
            LOG.error("Got one UnknownHostException when init client of id:{}", sortTaskName, ex);
        } catch (Throwable th) {
            LOG.error("Got one throwable when init client of id:{}", sortTaskName, th);
        }
        return null;
    }
Create one {@link SortClient} with specific sort task.

<p>
In current version, the {@link FetchCallback} will hold the client to ACK. For more details see
{@link FetchCallback#onFinished}
</p>

@param  sortTaskName Sort in of new client.
@return        New sort client.
inlong-sort-standalone/sort-standalone-source/src/main/java/org/apache/inlong/sort/standalone/source/sortsdk/SortSdkSource.java