@Override
    protected void addBatch(RowData in) throws Exception {
        checkError();
        try {
            if (running) {
                queue.put(in);
            } else {
                throw new FlinkRuntimeException("Canceled by KuduConsumerTask!");
            }
        } catch (InterruptedException e) {
            LOG.error("writeKuduMsgToQueue error", e);
        }
    }
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