You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/09/28 15:10:58 UTC
[GitHub] [beam] echauchot commented on a change in pull request #15381: [BEAM-10990] Elasticsearch response filtering
echauchot commented on a change in pull request #15381:
URL: https://github.com/apache/beam/pull/15381#discussion_r716701847
##########
File path: sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
##########
@@ -1623,11 +1651,153 @@ public void setup() throws IOException {
@ProcessElement
public void processElement(ProcessContext c) throws IOException {
- c.output(createBulkApiEntity(spec, c.element(), backendVersion));
+ String inputDoc = c.element();
+ String bulkDirective = createBulkApiEntity(spec, inputDoc, backendVersion);
+ c.output(
+ WriteSummary.create()
+ .withInputDoc(inputDoc)
+ .withBulkDirective(bulkDirective)
+ // N.B. Saving the element timestamp for later use allows for exactly emulating
+ // c.output(...) because c.output is equivalent to
+ // c.outputWithTimestamp(..., c.timestamp())
+ .withTimestamp(c.timestamp()));
}
}
}
+ public static class WriteSummaryCoder extends AtomicCoder<WriteSummary> implements Serializable {
+ private static final WriteSummaryCoder INSTANCE = new WriteSummaryCoder();
+
+ private WriteSummaryCoder() {}
+
+ public static WriteSummaryCoder of() {
+ return INSTANCE;
+ }
+
+ @Override
+ public void encode(WriteSummary value, OutputStream outStream) throws IOException {
+ NullableCoder.of(StringUtf8Coder.of()).encode(value.getInputDoc(), outStream);
+ NullableCoder.of(StringUtf8Coder.of()).encode(value.getBulkDirective(), outStream);
+ BooleanCoder.of().encode(value.getHasError(), outStream);
+ NullableCoder.of(StringUtf8Coder.of()).encode(value.getResponseItemJson(), outStream);
+ NullableCoder.of(InstantCoder.of()).encode(value.getTimestamp(), outStream);
+ }
+
+ @Override
+ public WriteSummary decode(InputStream inStream) throws IOException {
+ String inputDoc = NullableCoder.of(StringUtf8Coder.of()).decode(inStream);
+ String bulkDirective = NullableCoder.of(StringUtf8Coder.of()).decode(inStream);
+ boolean hasError = BooleanCoder.of().decode(inStream);
+ String responseItemJson = NullableCoder.of(StringUtf8Coder.of()).decode(inStream);
+ Instant timestamp = NullableCoder.of(InstantCoder.of()).decode(inStream);
+
+ return WriteSummary.create()
+ .withInputDoc(inputDoc)
+ .withBulkDirective(bulkDirective)
+ .withHasError(hasError)
+ .withResponseItemJson(responseItemJson)
+ .withTimestamp(timestamp);
+ }
+ }
+
+ // Immutable POJO for maintaining various states of documents and their bulk representation, plus
+ // response from ES for the given document and the timestamp of the data
+ @DefaultCoder(WriteSummaryCoder.class)
+ @AutoValue
+ public abstract static class WriteSummary implements Serializable {
Review comment:
As this class is instanciated before any write operation (in the DocToBulk transform) I'd call it simply `Document` and that would echo with `DocumentMetaData` existing class. WDYT ?
##########
File path: sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
##########
@@ -237,49 +250,64 @@ static JsonNode parseResponse(HttpEntity responseEntity) throws IOException {
return mapper.readValue(responseEntity.getContent(), JsonNode.class);
}
- static void checkForErrors(HttpEntity responseEntity, @Nullable Set<String> allowedErrorTypes)
+ static List<WriteSummary> checkForErrors(
+ HttpEntity responseEntity, @Nullable Set<String> allowedErrorTypes, boolean throwWriteErrors)
throws IOException {
+ List<WriteSummary> responses = new ArrayList<>();
+ int numErrors = 0;
JsonNode searchResult = parseResponse(responseEntity);
- boolean errors = searchResult.path("errors").asBoolean();
- if (errors) {
- int numErrors = 0;
-
- StringBuilder errorMessages =
- new StringBuilder("Error writing to Elasticsearch, some elements could not be inserted:");
- JsonNode items = searchResult.path("items");
- if (items.isMissingNode() || items.size() == 0) {
- errorMessages.append(searchResult.toString());
- }
- // some items present in bulk might have errors, concatenate error messages
- for (JsonNode item : items) {
- JsonNode error = item.findValue("error");
- if (error != null) {
- // N.B. An empty-string within the allowedErrorTypes Set implies all errors are allowed.
- String type = error.path("type").asText();
- String reason = error.path("reason").asText();
- String docId = item.findValue("_id").asText();
- JsonNode causedBy = error.path("caused_by"); // May not be present
- String cbReason = causedBy.path("reason").asText();
- String cbType = causedBy.path("type").asText();
-
- if (allowedErrorTypes == null
- || (!allowedErrorTypes.contains(type) && !allowedErrorTypes.contains(cbType))) {
- // 'error' and 'causedBy` fields are not null, and the error is not being ignored.
- numErrors++;
-
- errorMessages.append(String.format("%nDocument id %s: %s (%s)", docId, reason, type));
-
- if (!causedBy.isMissingNode()) {
- errorMessages.append(String.format("%nCaused by: %s (%s)", cbReason, cbType));
- }
+ StringBuilder errorMessages =
+ new StringBuilder("Error writing to Elasticsearch, some elements could not be inserted:");
+ JsonNode items = searchResult.path("items");
+
+ if (items.isMissingNode() || items.size() == 0) {
+ // This would only be expected in cases like connectivity issues or similar
+ errorMessages.append(searchResult.toString());
+ throw new RuntimeException(
+ String.format(
+ "'items' missing from Elasticsearch response: %s", errorMessages.toString()));
+ }
+
+ // some items present in bulk might have errors, concatenate error messages and record
+ // which items had errors
+ for (JsonNode item : items) {
+ WriteSummary result = WriteSummary.create().withResponseItemJson(item.toString());
+
+ JsonNode error = item.findValue("error");
+ if (error != null) {
+ // N.B. An empty-string within the allowedErrorTypes Set implies all errors are allowed.
+ String type = error.path("type").asText();
+ String reason = error.path("reason").asText();
+ String docId = item.findValue("_id").asText();
+ JsonNode causedBy = error.path("caused_by"); // May not be present
+ String cbReason = causedBy.path("reason").asText();
+ String cbType = causedBy.path("type").asText();
+
+ if (allowedErrorTypes == null
+ || (!allowedErrorTypes.contains(type) && !allowedErrorTypes.contains(cbType))) {
+ // 'error' and 'causedBy` fields are not null, and the error is not being ignored.
+ result = result.withHasError(true);
+ numErrors++;
+
+ errorMessages.append(String.format("%nDocument id %s: %s (%s)", docId, reason, type));
+
+ if (!causedBy.isMissingNode()) {
+ errorMessages.append(String.format("%nCaused by: %s (%s)", cbReason, cbType));
}
}
}
- if (numErrors > 0) {
+ responses.add(result);
+ }
+
+ if (numErrors > 0) {
+ LOG.error(errorMessages.toString());
+ if (throwWriteErrors) {
Review comment:
l like that this behavior is configurable !
##########
File path: sdks/java/io/elasticsearch-tests/elasticsearch-tests-5/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOIT.java
##########
@@ -141,6 +141,18 @@ public void testWriteWithAllowableErrors() throws Exception {
elasticsearchIOTestCommon.testWriteWithAllowedErrors();
}
+ @Test
Review comment:
No need to test in ITest it is more of an UTest. ITests are for high load
##########
File path: sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
##########
@@ -237,49 +250,64 @@ static JsonNode parseResponse(HttpEntity responseEntity) throws IOException {
return mapper.readValue(responseEntity.getContent(), JsonNode.class);
}
- static void checkForErrors(HttpEntity responseEntity, @Nullable Set<String> allowedErrorTypes)
+ static List<WriteSummary> checkForErrors(
+ HttpEntity responseEntity, @Nullable Set<String> allowedErrorTypes, boolean throwWriteErrors)
throws IOException {
+ List<WriteSummary> responses = new ArrayList<>();
+ int numErrors = 0;
JsonNode searchResult = parseResponse(responseEntity);
- boolean errors = searchResult.path("errors").asBoolean();
- if (errors) {
- int numErrors = 0;
-
- StringBuilder errorMessages =
- new StringBuilder("Error writing to Elasticsearch, some elements could not be inserted:");
- JsonNode items = searchResult.path("items");
- if (items.isMissingNode() || items.size() == 0) {
- errorMessages.append(searchResult.toString());
- }
- // some items present in bulk might have errors, concatenate error messages
- for (JsonNode item : items) {
- JsonNode error = item.findValue("error");
- if (error != null) {
- // N.B. An empty-string within the allowedErrorTypes Set implies all errors are allowed.
- String type = error.path("type").asText();
- String reason = error.path("reason").asText();
- String docId = item.findValue("_id").asText();
- JsonNode causedBy = error.path("caused_by"); // May not be present
- String cbReason = causedBy.path("reason").asText();
- String cbType = causedBy.path("type").asText();
-
- if (allowedErrorTypes == null
- || (!allowedErrorTypes.contains(type) && !allowedErrorTypes.contains(cbType))) {
- // 'error' and 'causedBy` fields are not null, and the error is not being ignored.
- numErrors++;
-
- errorMessages.append(String.format("%nDocument id %s: %s (%s)", docId, reason, type));
-
- if (!causedBy.isMissingNode()) {
- errorMessages.append(String.format("%nCaused by: %s (%s)", cbReason, cbType));
- }
+ StringBuilder errorMessages =
+ new StringBuilder("Error writing to Elasticsearch, some elements could not be inserted:");
+ JsonNode items = searchResult.path("items");
+
+ if (items.isMissingNode() || items.size() == 0) {
+ // This would only be expected in cases like connectivity issues or similar
+ errorMessages.append(searchResult.toString());
+ throw new RuntimeException(
Review comment:
fail the pipeline in that case ?
##########
File path: sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
##########
@@ -1789,8 +1964,14 @@ public Write withAllowableResponseErrors(@Nullable Set<String> allowableResponse
return this;
}
+ /** Refer to {@link BulkIO#withThrowWriteErrors}. */
+ public Write withThrowWriteErrors(boolean throwWriteErrors) {
Review comment:
Please add a hint to tell users that this parameter could be usefull in case the pipeline is in streaming mode to avoid infinite retry
##########
File path: sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
##########
@@ -1636,11 +1806,15 @@ public void processElement(ProcessContext c) throws IOException {
* cluster. This class is effectively a thin proxy for DocToBulk->BulkIO all-in-one for
* convenience and backward compatibility.
*/
- public static class Write extends PTransform<PCollection<String>, PDone> {
+ public static class Write extends PTransform<PCollection<String>, PCollectionTuple> {
Review comment:
I see the write tests do not change so there is no breaking change. I guess the user could chose whether he wants to process the PCollectionTuple or not.
##########
File path: sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
##########
@@ -237,49 +250,64 @@ static JsonNode parseResponse(HttpEntity responseEntity) throws IOException {
return mapper.readValue(responseEntity.getContent(), JsonNode.class);
}
- static void checkForErrors(HttpEntity responseEntity, @Nullable Set<String> allowedErrorTypes)
+ static List<WriteSummary> checkForErrors(
Review comment:
I would rename this method. Indeed, it can also report items written with sucess so I'd call it something like `createWriteReport`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org