public void invoke(RowData in, Context context) {

        List<OUT> redisOutputs = serialize(in);
        sendMetrics(in.toString().getBytes());
        synchronized (lock) {
            rows.addAll(redisOutputs);
            if (forceFlush || rows.size() >= batchSize) {
                flush();
            }
        }

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