@Override
		public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
			if (response.hasFailures()) {
				BulkItemResponse itemResponse;
				Throwable failure;
				RestStatus restStatus;

				try {
					for (int i = 0; i < response.getItems().length; i++) {
						itemResponse = response.getItems()[i];
						failure = callBridge.extractFailureCauseFromBulkItemResponse(itemResponse);
						if (failure != null) {
							LOG.error("Failed Elasticsearch item request: {}", itemResponse.getFailureMessage(), failure);

							restStatus = itemResponse.getFailure().getStatus();
							if (restStatus == null) {
								failureHandler.onFailure((ActionRequest) request.requests().get(i), failure, -1, failureRequestIndexer);
							} else {
								failureHandler.onFailure((ActionRequest) request.requests().get(i), failure, restStatus.getStatus(), failureRequestIndexer);
							}
						}
					}
				} catch (Throwable t) {
					// fail the sink and skip the rest of the items
					// if the failure handler decides to throw an exception
					failureThrowable.compareAndSet(null, t);
				}
			}

			if (flushOnCheckpoint) {
				numPendingRequests.getAndAdd(-request.numberOfActions());
			}
		}
The interval time which check client whether available.
elasticsearch5-xh/elasticsearch5-xh-sink/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java