@Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        ActionRequestFailureHandler<?> failureHandler = getFailureHandler();
        if (failureHandler instanceof DirtySinkFailureHandler) {
            ((DirtySinkFailureHandler) failureHandler).setSinkMetricData(getSinkMetricData());
            ((DirtySinkFailureHandler) failureHandler).setDirtySinkHelper(getDirtySinkHelper());
        }
    }
Elasticsearch 7.x sink that requests multiple {@link ActionRequest ActionRequests} against a
cluster for each incoming element.

<p>The sink internally uses a {@link RestHighLevelClient} to communicate with an Elasticsearch
cluster. The sink will fail if no cluster can be connected to using the provided transport
addresses passed to the constructor.

<p>Internally, the sink will use a {@link BulkProcessor} to send {@link ActionRequest
ActionRequests}. This will buffer elements before sending a request to the cluster. The behaviour
of the {@code BulkProcessor} can be configured using these config keys:

<ul>
  <li>{@code bulk.flush.max.actions}: Maximum amount of elements to buffer
  <li>{@code bulk.flush.max.size.mb}: Maximum amount of data (in megabytes) to buffer
  <li>{@code bulk.flush.interval.ms}: Interval at which to flush data regardless of the other two
      settings in milliseconds
</ul>

<p>You also have to provide an {@link ElasticsearchSinkFunction}. This is used to create multiple
{@link ActionRequest ActionRequests} for each incoming element. See the class level documentation
of {@link ElasticsearchSinkFunction} for an example.

@param <T> Type of the elements handled by this sink
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/elasticsearch-7/src/main/java/org/apache/inlong/sort/elasticsearch7/ElasticsearchSink.java