You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2019/02/11 14:14:24 UTC
[flink] 02/04: [FLINK-11046] [elasticsearch] Simplify failure
handler indexer implementation
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch release-1.7
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 35af99391dac431c85e30bcc98b89cba79bccfea
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Sat Feb 9 18:11:25 2019 +0800
[FLINK-11046] [elasticsearch] Simplify failure handler indexer implementation
Since we always use the original request indexer to re-index failed
requests, and that indexer already keeps track of the number of pending
actions, the ElasticsearchFailureHandlerIndexer does not need to keep
track of the number of pending records by itself.
---
.../elasticsearch/BufferingNoOpRequestIndexer.java | 76 +++++++++++++++++++
.../elasticsearch/ElasticsearchApiCallBridge.java | 12 ---
.../ElasticsearchFailureHandlerIndexer.java | 88 ----------------------
.../elasticsearch/ElasticsearchSinkBase.java | 20 ++---
4 files changed, 82 insertions(+), 114 deletions(-)
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/BufferingNoOpRequestIndexer.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/BufferingNoOpRequestIndexer.java
new file mode 100644
index 0000000..e639b82
--- /dev/null
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/BufferingNoOpRequestIndexer.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.connectors.elasticsearch;
+
+import org.apache.flink.annotation.Internal;
+
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.delete.DeleteRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.action.update.UpdateRequest;
+
+import javax.annotation.concurrent.NotThreadSafe;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+/**
+ * Implementation of a {@link RequestIndexer} that buffers {@link ActionRequest ActionRequests}
+ * before re-sending them to the Elasticsearch cluster upon request.
+ */
+@Internal
+@NotThreadSafe
+class BufferingNoOpRequestIndexer implements RequestIndexer {
+
+ private List<ActionRequest> bufferedRequests;
+
+ BufferingNoOpRequestIndexer() {
+ this.bufferedRequests = new ArrayList<>(10);
+ }
+
+ @Override
+ public void add(DeleteRequest... deleteRequests) {
+ Collections.addAll(bufferedRequests, deleteRequests);
+ }
+
+ @Override
+ public void add(IndexRequest... indexRequests) {
+ Collections.addAll(bufferedRequests, indexRequests);
+ }
+
+ @Override
+ public void add(UpdateRequest... updateRequests) {
+ Collections.addAll(bufferedRequests, updateRequests);
+ }
+
+ void processBufferedRequests(RequestIndexer actualIndexer) {
+ for (ActionRequest request : bufferedRequests) {
+ if (request instanceof IndexRequest) {
+ actualIndexer.add((IndexRequest) request);
+ } else if (request instanceof DeleteRequest) {
+ actualIndexer.add((DeleteRequest) request);
+ } else if (request instanceof UpdateRequest) {
+ actualIndexer.add((UpdateRequest) request);
+ }
+ }
+
+ bufferedRequests.clear();
+ }
+}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchApiCallBridge.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchApiCallBridge.java
index e450485..d3b774c 100644
--- a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchApiCallBridge.java
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchApiCallBridge.java
@@ -94,18 +94,6 @@ public interface ElasticsearchApiCallBridge<C extends AutoCloseable> extends Ser
}
/**
- * Creates a {@link RequestIndexer} that is able to work with {@link BulkProcessor} binary compatible.
- */
- default ElasticsearchFailureHandlerIndexer createFailureHandlerIndexer(
- boolean flushOnCheckpoint,
- AtomicLong numPendingRequestsRef) {
- return new ElasticsearchFailureHandlerIndexer(
- flushOnCheckpoint,
- numPendingRequestsRef);
- }
-
-
- /**
* Perform any necessary state cleanup.
*/
default void cleanup() {
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchFailureHandlerIndexer.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchFailureHandlerIndexer.java
deleted file mode 100644
index 685b5d9..0000000
--- a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchFailureHandlerIndexer.java
+++ /dev/null
@@ -1,88 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.flink.streaming.connectors.elasticsearch;
-
-import org.apache.flink.annotation.Internal;
-
-import org.elasticsearch.action.ActionRequest;
-import org.elasticsearch.action.bulk.BulkRequest;
-import org.elasticsearch.action.delete.DeleteRequest;
-import org.elasticsearch.action.index.IndexRequest;
-import org.elasticsearch.action.update.UpdateRequest;
-
-import java.util.concurrent.atomic.AtomicLong;
-
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/**
- * Implementation of a {@link RequestIndexer}, using a {@link BulkRequest}.
- * {@link ActionRequest ActionRequests} will be buffered before re-sending a bulk request to the Elasticsearch cluster.
- */
-
-@Internal
-class ElasticsearchFailureHandlerIndexer implements RequestIndexer {
-
- private BulkRequest bulkRequest;
- private final boolean flushOnCheckpoint;
- private final AtomicLong numPendingRequestsRef;
-
- ElasticsearchFailureHandlerIndexer(boolean flushOnCheckpoint, AtomicLong numPendingRequestsRef) {
- this.bulkRequest = new BulkRequest();
- this.flushOnCheckpoint = flushOnCheckpoint;
- this.numPendingRequestsRef = checkNotNull(numPendingRequestsRef);
- }
-
- @Override
- public void add(DeleteRequest... deleteRequests) {
- for (DeleteRequest deleteRequest : deleteRequests) {
- if (flushOnCheckpoint) {
- numPendingRequestsRef.getAndIncrement();
- }
- this.bulkRequest.add(deleteRequest);
- }
- }
-
- @Override
- public void add(IndexRequest... indexRequests) {
- for (IndexRequest indexRequest : indexRequests) {
- if (flushOnCheckpoint) {
- numPendingRequestsRef.getAndIncrement();
- }
- this.bulkRequest.add(indexRequest);
- }
- }
-
- @Override
- public void add(UpdateRequest... updateRequests) {
- for (UpdateRequest updateRequest : updateRequests) {
- if (flushOnCheckpoint) {
- numPendingRequestsRef.getAndIncrement();
- }
- this.bulkRequest.add(updateRequest);
- }
- }
-
- public BulkRequest getBulkRequest() {
- return bulkRequest;
- }
-
- public int numberOfActions() {
- return bulkRequest.numberOfActions();
- }
-}
diff --git a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java
index 55d8854..e50140d 100644
--- a/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java
+++ b/flink-connectors/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java
@@ -166,7 +166,8 @@ public abstract class ElasticsearchSinkBase<T, C extends AutoCloseable> extends
/** Provided to the user via the {@link ElasticsearchSinkFunction} to add {@link ActionRequest ActionRequests}. */
private transient RequestIndexer requestIndexer;
- private transient ElasticsearchFailureHandlerIndexer failureRequestIndexer;
+ /** Provided to the {@link ActionRequestFailureHandler} to allow users to re-index failed requests. */
+ private transient BufferingNoOpRequestIndexer failureRequestIndexer;
// ------------------------------------------------------------------------
// Internals for the Flink Elasticsearch Sink
@@ -298,15 +299,15 @@ public abstract class ElasticsearchSinkBase<T, C extends AutoCloseable> extends
client = callBridge.createClient(userConfig);
bulkProcessor = buildBulkProcessor(new BulkProcessorListener());
requestIndexer = callBridge.createBulkProcessorIndexer(bulkProcessor, flushOnCheckpoint, numPendingRequests);
- failureRequestIndexer = callBridge.createFailureHandlerIndexer(flushOnCheckpoint, numPendingRequests);
+ failureRequestIndexer = new BufferingNoOpRequestIndexer();
}
@Override
public void invoke(T value) throws Exception {
// if bulk processor callbacks have previously reported an error, we rethrow the error and fail the sink
checkErrorAndRethrow();
- reindexFailedRequest();
+ failureRequestIndexer.processBufferedRequests(requestIndexer);
elasticsearchSinkFunction.process(value, getRuntimeContext(), requestIndexer);
}
@@ -320,10 +321,10 @@ public abstract class ElasticsearchSinkBase<T, C extends AutoCloseable> extends
checkErrorAndRethrow();
if (flushOnCheckpoint) {
- do {
+ while (numPendingRequests.get() != 0) {
bulkProcessor.flush();
checkErrorAndRethrow();
- } while (numPendingRequests.get() != 0);
+ }
}
}
@@ -384,15 +385,6 @@ public abstract class ElasticsearchSinkBase<T, C extends AutoCloseable> extends
}
}
- private void reindexFailedRequest() {
- if (failureRequestIndexer.numberOfActions() > 0) {
- BulkRequest failedRequest = failureRequestIndexer.getBulkRequest();
- for (ActionRequest request: failedRequest.requests()) {
- requestIndexer.add(request);
- }
- }
- }
-
private class BulkProcessorListener implements BulkProcessor.Listener {
@Override
public void beforeBulk(long executionId, BulkRequest request) { }