Apache Flink Elasticsearch 6 Connector

import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.ActionRequestFailureHandler;
import org.apache.flink.streaming.connectors.elasticsearch6.RestClientFactory;
import org.apache.http.HttpHost;

// Configuration via DDL
CREATE TABLE elasticsearch_sink (
  id BIGINT,
  name STRING,
  age INT
) WITH (
  'connector' = 'elasticsearch-6',
  'hosts' = 'http://localhost:9200',
  'index' = 'users',
  'document-type' = '_doc'
);

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink;
import org.apache.flink.streaming.connectors.elasticsearch.ElasticsearchSinkFunction;
import org.apache.flink.streaming.connectors.elasticsearch.RequestIndexer;
import org.apache.http.HttpHost;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.Requests;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

// Create HTTP hosts list
List<HttpHost> httpHosts = new ArrayList<>();
httpHosts.add(new HttpHost("127.0.0.1", 9200, "http"));

// Define sink function
ElasticsearchSinkFunction<String> sinkFunction = new ElasticsearchSinkFunction<String>() {
    public IndexRequest createIndexRequest(String element) {
        Map<String, Object> json = new HashMap<>();
        json.put("data", element);
        
        return Requests.indexRequest()
                .index("my-index")
                .type("_doc")
                .source(json);
    }
    
    @Override
    public void process(String element, RuntimeContext ctx, RequestIndexer indexer) {
        indexer.add(createIndexRequest(element));
    }
};

// Create sink
ElasticsearchSink<String> sink = new ElasticsearchSink.Builder<>(
    httpHosts,
    sinkFunction
).build();

// Add sink to DataStream
DataStream<String> input = // ... your data stream
input.addSink(sink);

public class ElasticsearchSink<T> extends ElasticsearchSinkBase<T, RestHighLevelClient> {
    // Private constructor - use Builder
}

public static class ElasticsearchSink.Builder<T> {
    public Builder(List<HttpHost> httpHosts, ElasticsearchSinkFunction<T> elasticsearchSinkFunction);
    public ElasticsearchSink<T> build();
}

// Table API usage via DDL
CREATE TABLE sink_table (...) WITH (
  'connector' = 'elasticsearch-6',
  'hosts' = 'http://localhost:9200',
  'index' = 'target-index',
  'document-type' = '_doc'
);

// Builder methods for bulk configuration
public void setBulkFlushMaxActions(int numMaxActions);
public void setBulkFlushMaxSizeMb(int maxSizeMb);
public void setBulkFlushInterval(long intervalMillis);

public interface ActionRequestFailureHandler extends Serializable {
    void onFailure(
        ActionRequest action, 
        Throwable failure, 
        int restStatusCode, 
        RequestIndexer indexer
    ) throws Throwable;
}

public interface RestClientFactory extends Serializable {
    void configureRestClientBuilder(RestClientBuilder restClientBuilder);
}

// Core functional interface for processing stream elements
public interface ElasticsearchSinkFunction<T> extends Serializable, Function {
    default void open() throws Exception {}
    default void close() throws Exception {}
    void process(T element, RuntimeContext ctx, RequestIndexer indexer);
}

// Request indexer for adding ActionRequests
public interface RequestIndexer {
    void add(DeleteRequest... deleteRequests);
    void add(IndexRequest... indexRequests); 
    void add(UpdateRequest... updateRequests);
}

// Backoff configuration
public enum FlushBackoffType {
    CONSTANT,
    EXPONENTIAL
}

// Backoff configuration policy for bulk processing
public static class BulkFlushBackoffPolicy implements Serializable {
    /**
     * Get the backoff type (CONSTANT or EXPONENTIAL).
     * @return the backoff type
     */
    public FlushBackoffType getBackoffType();
    
    /**
     * Get the maximum number of retry attempts.
     * @return the maximum retry count
     */
    public int getMaxRetryCount();
    
    /**
     * Get the initial delay in milliseconds.
     * @return the delay in milliseconds
     */
    public long getDelayMillis();
    
    /**
     * Set the backoff type.
     * @param backoffType the backoff type to use
     */
    public void setBackoffType(FlushBackoffType backoffType);
    
    /**
     * Set the maximum number of retry attempts.
     * @param maxRetryCount the maximum retry count (must be >= 0)
     */
    public void setMaxRetryCount(int maxRetryCount);
    
    /**
     * Set the initial delay between retry attempts.
     * @param delayMillis the delay in milliseconds (must be >= 0)
     */
    public void setDelayMillis(long delayMillis);
}