@Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        int threadCnt = configuration.getInteger(WRITE_THREAD_COUNT);
        int cacheQueueMaxLength = configuration.getInteger(CACHE_QUEUE_MAX_LENGTH);
        if (cacheQueueMaxLength == -1) {
            cacheQueueMaxLength = (maxBufferSize + 1) * (threadCnt + 1);
        }
        LOG.info("Opening KuduAsyncSinkFunction, threadCount:{}, cacheQueueMaxLength:{}, maxBufferSize:{}.",
                threadCnt, cacheQueueMaxLength, maxBufferSize);

        queue = new LinkedBlockingQueue<>(cacheQueueMaxLength);
        consumerTasks = new ArrayList<>(threadCnt);
        ThreadFactory namedThreadFactory = new ThreadFactoryBuilder()
                .setNameFormat("kudu-sink-pool-%d").build();
        threadPool = new ThreadPoolExecutor(
                threadCnt + 1,
                threadCnt + 1,
                0L,
                TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(4),
                namedThreadFactory,
                new ThreadPoolExecutor.AbortPolicy());

        KuduClient client = buildKuduClient();
        KuduTable kuduTable;
        try {
            String tableName = kuduTableInfo.getTableName();
            kuduTable = client.openTable(tableName);
            checkState(client.tableExists(tableName), "Can not find table with name:{} in kudu.", tableName);
        } catch (KuduException e) {
            LOG.error("Error on Open kudu table", e);
            throw new RuntimeException(e);
        }

        // check table schema
        Schema schema = kuduTable.getSchema();
        try {
            checkSchema(kuduTableInfo.getFieldNames(), kuduTableInfo.getDataTypes(), schema);
        } catch (Exception e) {
            LOG.error("The provided schema is invalid!", e);
            throw new RuntimeException(e);
        }
        boolean forceInUpsertMode = configuration.getBoolean(SINK_FORCE_WITH_UPSERT_MODE);
        for (int threadIndex = 0; threadIndex < threadCnt; threadIndex++) {
            KuduWriter kuduWriter = new KuduWriter(client, kuduTable, kuduTableInfo);
            KuduConsumerTask task = new KuduConsumerTask(
                    queue, kuduWriter, maxBufferSize, maxBufferTime, maxRetries);
            consumerTasks.add(task);
            threadPool.submit(task);
        }
    }
The Flink kudu Producer in async Mode.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kudu/src/main/java/org/apache/inlong/sort/kudu/sink/KuduAsyncSinkFunction.java