@Override
    public void open(Configuration parameters) throws Exception {
        client = callBridge.createClient(userConfig);
        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(RegisteredMetric.ALL)
                .build();
        if (metricOption != null) {
            sinkMetricData = new SinkTableMetricData(metricOption, getRuntimeContext().getMetricGroup());
            if (multipleSink) {
                sinkMetricData.registerSubMetricsGroup(metricState);
            }
        }
        dirtySinkHelper.open(parameters);
        Listener listener = createListener();
        bulkProcessor = buildBulkProcessor(listener);
        requestIndexer = callBridge.createBulkProcessorIndexer(bulkProcessor, flushOnCheckpoint, numPendingRequests);
        callBridge.verifyClientConnection(client);
        failureRequestIndexer = new BufferingNoOpRequestIndexer<>();
        elasticsearchSinkFunction.open(getRuntimeContext(), sinkMetricData);
    }
Disable flushing on checkpoint. When disabled, the sink will not wait for all pending action
requests to be acknowledged by Elasticsearch on checkpoints.

<p>NOTE: If flushing on checkpoint is disabled, the Flink Elasticsearch Sink does NOT provide
any strong guarantees for at-least-once delivery of action requests.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-base/src/main/java/org/apache/inlong/sort/elasticsearch/ElasticsearchSinkBase.java