@Override
    public void run() {
        LOG.info("KuduConsumerTask starting: {}.", Thread.currentThread());
        try {
            while (running) {
                // Data flush conditions.
                while (running && cachedNum < writeBatch) {
                    RowData row = null;
                    try {
                        row = queue.poll(maxBufferTime.getSeconds(), TimeUnit.SECONDS);
                    } catch (InterruptedException e) {
                        LOG.error("Poll message error, force flush", e);
                        flush();
                    }
                    if (row != null) {
                        applyRow(row);
                    }
                }
                flush();
                try {
                    Thread.sleep(1);
                } catch (InterruptedException ignored) {
                }
            }
        } catch (Exception e) {
            LOG.error("", e);
        }
        LOG.info("KuduConsumerTask stopped: {}.", Thread.currentThread());
    }
The maximum number of retries when an exception is caught.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/kudu/src/main/java/org/apache/inlong/sort/kudu/source/KuduConsumerTask.java