Flink写Elasticsearch导致Checkpoint频繁失败的解决方案

public abstract class ElasticsearchSinkBase<T, C extends AutoCloseable> extends RichSinkFunction<T> implements CheckpointedFunction {

 /** If true, the producer will wait until all outstanding action requests have been sent to Elasticsearch. */
 private boolean flushOnCheckpoint = true;

 /**
  * 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.
  */
 public void disableFlushOnCheckpoint() {
  this.flushOnCheckpoint = false;
 }

 @Override
 public void snapshotState(FunctionSnapshotContext context) throws Exception {
  checkAsyncErrorsAndRequests();

  if (flushOnCheckpoint) {
   while (numPendingRequests.get() != 0) {
    bulkProcessor.flush();
    checkAsyncErrorsAndRequests();
   }
  }
 }
}

ElasticsearchSink<Row> elasticsearchSink = esSinkBuilder.build();
// 关掉基于Checkpoint的flush
elasticsearchSink.disableFlushOnCheckpoint();