@Override
    public void open(Configuration parameters) {
        LOG.info("Opening redis sink with address");

        lock = new Object();

        stopWatch = new StopWatch();

        try {
            this.redisCommandsContainer = RedisCommandsContainerBuilder.build(this.flinkJedisConfigBase);
            this.redisCommandsContainer.open();
        } catch (Exception e) {
            LOG.error("Redis has not been properly initialized: ", e);
            throw new RuntimeException(e);
        }

        Optional<OutputFlusher> outputFlusher;
        if (this.batchSize == 1 || this.flushIntervalInMillis == 0) {
            LOG.info("Flush records immediately.");
            outputFlusher = Optional.empty();
        } else {
            String threadName = DEFAULT_OUTPUT_FLUSH_THREAD_NAME + " for "
                    + getRuntimeContext().getTaskNameWithSubtasks();
            outputFlusher = Optional.of(new OutputFlusher(threadName, flushIntervalInMillis));
            outputFlusher.get().start();
        }
        MetricOption metricOption = MetricOption.builder()
                .withInlongLabels(inLongMetric)
                .withAuditAddress(auditHostAndPorts)
                .withInitRecords(metricState != null ? metricState.getMetricValue(NUM_RECORDS_OUT) : 0L)
                .withInitBytes(metricState != null ? metricState.getMetricValue(NUM_BYTES_OUT) : 0L)
                .withInitDirtyRecords(metricState != null ? metricState.getMetricValue(DIRTY_RECORDS_OUT) : 0L)
                .withInitDirtyBytes(metricState != null ? metricState.getMetricValue(DIRTY_BYTES_OUT) : 0L)
                .withRegisterMetric(MetricOption.RegisteredMetric.ALL)
                .build();
        if (metricOption != null) {
            sinkMetricData = new SinkMetricData(metricOption, getRuntimeContext().getMetricGroup());
        }
    }
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