@Override
        public void run() {
            while (running) {

                try {
                    try {
                        Thread.sleep(timeoutInMillis);
                    } catch (InterruptedException e) {
                        if (running) {
                            throw new Exception(e);
                        }
                    }
                    if (rows.size() > 0) {
                        flush();
                    }
                } catch (Throwable t) {
                    LOG.error("An exception happened while flushing the outputs", t);
                    // There is no need to handle exceptions in asynchronous threads.
                    // When the number of rows exceeds the batchSize, it will fail directly in the next write.
                    // But there is a possibility of data delay.
                    forceFlush = true;
                    LOG.error("Set the forceFlush to true, it will retry in the main thread.");
                }

            }
        }
The stop watch to measure time duration.
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/redis/src/main/java/org/apache/inlong/sort/redis/sink/AbstractRedisSinkFunction.java