You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ec...@apache.org on 2022/01/19 09:19:44 UTC
[beam] branch master updated: Merge pull request #16309: [BEAM-13503] Set a default value to throwWriteErrors in BulkIO constructor
This is an automated email from the ASF dual-hosted git repository.
echauchot pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new fd3f207 Merge pull request #16309: [BEAM-13503] Set a default value to throwWriteErrors in BulkIO constructor
fd3f207 is described below
commit fd3f2073bfc91ab866d97ae2b3591261212be9e7
Author: Kyle Hersey <km...@gmail.com>
AuthorDate: Wed Jan 19 04:18:19 2022 -0500
Merge pull request #16309: [BEAM-13503] Set a default value to throwWriteErrors in BulkIO constructor
[BEAM-13503] Set a default value to throwWriteErrors in BulkIO constructor
---
.../sdk/io/elasticsearch/ElasticsearchIOTest.java | 6 ++++++
.../sdk/io/elasticsearch/ElasticsearchIOTest.java | 6 ++++++
.../sdk/io/elasticsearch/ElasticsearchIOTest.java | 6 ++++++
.../io/elasticsearch/ElasticsearchIOTestCommon.java | 19 +++++++++++++++++++
.../beam/sdk/io/elasticsearch/ElasticsearchIO.java | 1 +
5 files changed, 38 insertions(+)
diff --git a/sdks/java/io/elasticsearch-tests/elasticsearch-tests-5/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTest.java b/sdks/java/io/elasticsearch-tests/elasticsearch-tests-5/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTest.java
index 4023b24..6dd2c69 100644
--- a/sdks/java/io/elasticsearch-tests/elasticsearch-tests-5/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTest.java
+++ b/sdks/java/io/elasticsearch-tests/elasticsearch-tests-5/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTest.java
@@ -265,6 +265,12 @@ public class ElasticsearchIOTest implements Serializable {
}
@Test
+ public void testDocToBulkAndBulkIO() throws Exception {
+ elasticsearchIOTestCommon.setPipeline(pipeline);
+ elasticsearchIOTestCommon.testDocToBulkAndBulkIO();
+ }
+
+ @Test
public void testDocumentCoder() throws Exception {
elasticsearchIOTestCommon.testDocumentCoder();
}
diff --git a/sdks/java/io/elasticsearch-tests/elasticsearch-tests-6/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTest.java b/sdks/java/io/elasticsearch-tests/elasticsearch-tests-6/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTest.java
index 3e77be8..9b74161 100644
--- a/sdks/java/io/elasticsearch-tests/elasticsearch-tests-6/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTest.java
+++ b/sdks/java/io/elasticsearch-tests/elasticsearch-tests-6/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTest.java
@@ -258,6 +258,12 @@ public class ElasticsearchIOTest implements Serializable {
}
@Test
+ public void testDocToBulkAndBulkIO() throws Exception {
+ elasticsearchIOTestCommon.setPipeline(pipeline);
+ elasticsearchIOTestCommon.testDocToBulkAndBulkIO();
+ }
+
+ @Test
public void testDocumentCoder() throws Exception {
elasticsearchIOTestCommon.testDocumentCoder();
}
diff --git a/sdks/java/io/elasticsearch-tests/elasticsearch-tests-7/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTest.java b/sdks/java/io/elasticsearch-tests/elasticsearch-tests-7/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTest.java
index 4d78666..094e3e4 100644
--- a/sdks/java/io/elasticsearch-tests/elasticsearch-tests-7/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTest.java
+++ b/sdks/java/io/elasticsearch-tests/elasticsearch-tests-7/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTest.java
@@ -259,6 +259,12 @@ public class ElasticsearchIOTest implements Serializable {
}
@Test
+ public void testDocToBulkAndBulkIO() throws Exception {
+ elasticsearchIOTestCommon.setPipeline(pipeline);
+ elasticsearchIOTestCommon.testDocToBulkAndBulkIO();
+ }
+
+ @Test
public void testDocumentCoder() throws Exception {
elasticsearchIOTestCommon.testDocumentCoder();
}
diff --git a/sdks/java/io/elasticsearch-tests/elasticsearch-tests-common/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTestCommon.java b/sdks/java/io/elasticsearch-tests/elasticsearch-tests-common/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTestCommon.java
index 9bdcaee..981a151 100644
--- a/sdks/java/io/elasticsearch-tests/elasticsearch-tests-common/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTestCommon.java
+++ b/sdks/java/io/elasticsearch-tests/elasticsearch-tests-common/src/test/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIOTestCommon.java
@@ -287,6 +287,25 @@ class ElasticsearchIOTestCommon implements Serializable {
executeWriteTest(write);
}
+ /** Test that DocToBulk and BulkIO can be constructed and operate independently of Write. */
+ void testDocToBulkAndBulkIO() throws Exception {
+ DocToBulk docToBulk =
+ ElasticsearchIO.docToBulk().withConnectionConfiguration(connectionConfiguration);
+ BulkIO bulkIO = ElasticsearchIO.bulkIO().withConnectionConfiguration(connectionConfiguration);
+
+ List<String> data =
+ ElasticsearchIOTestUtils.createDocuments(
+ numDocs, ElasticsearchIOTestUtils.InjectionMode.DO_NOT_INJECT_INVALID_DOCS);
+ pipeline.apply(Create.of(data)).apply(docToBulk).apply(bulkIO);
+ pipeline.run();
+
+ long currentNumDocs = refreshIndexAndGetCurrentNumDocs(connectionConfiguration, restClient);
+ assertEquals(numDocs, currentNumDocs);
+
+ int count = countByScientistName(connectionConfiguration, restClient, "Einstein", null);
+ assertEquals(numDocs / NUM_SCIENTISTS, count);
+ }
+
void testWriteStateful() throws Exception {
Write write =
ElasticsearchIO.write()
diff --git a/sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java b/sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
index 33b39db..58142de 100644
--- a/sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
+++ b/sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java
@@ -234,6 +234,7 @@ public class ElasticsearchIO {
.setMaxBatchSizeBytes(5L * 1024L * 1024L)
.setUseStatefulBatches(false)
.setMaxParallelRequestsPerWindow(1)
+ .setThrowWriteErrors(true)
.build();
}