Repository: beam Updated Branches: refs/heads/master 16f8000e2 -> 6573e2ea5
[BEAM-2488] Elasticsearch IO should read also in replica shards Project: http://git-wip-us.apache.org/repos/asf/beam/repo Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/b1ed9757 Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/b1ed9757 Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/b1ed9757 Branch: refs/heads/master Commit: b1ed9757cead18b006d2e22c73fe1399a3022ae5 Parents: 16f8000 Author: Etienne Chauchot <[email protected]> Authored: Wed Jun 21 10:14:08 2017 +0200 Committer: Jean-Baptiste Onofré <[email protected]> Committed: Wed Jun 28 09:47:44 2017 +0200 ---------------------------------------------------------------------- sdks/java/io/elasticsearch/pom.xml | 8 ++++++++ .../beam/sdk/io/elasticsearch/ElasticsearchIO.java | 11 +---------- 2 files changed, 9 insertions(+), 10 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/beam/blob/b1ed9757/sdks/java/io/elasticsearch/pom.xml ---------------------------------------------------------------------- diff --git a/sdks/java/io/elasticsearch/pom.xml b/sdks/java/io/elasticsearch/pom.xml index 03632ce..c8e308c 100644 --- a/sdks/java/io/elasticsearch/pom.xml +++ b/sdks/java/io/elasticsearch/pom.xml @@ -137,6 +137,14 @@ <scope>test</scope> </dependency> + <!-- This optional dependency is used by the test framework. Avoids a warning --> + <dependency> + <groupId>net.java.dev.jna</groupId> + <artifactId>jna</artifactId> + <version>4.1.0</version> + <scope>test</scope> + </dependency> + <dependency> <groupId>org.apache.beam</groupId> <artifactId>beam-runners-direct-java</artifactId> http://git-wip-us.apache.org/repos/asf/beam/blob/b1ed9757/sdks/java/io/elasticsearch/src/main/java/org/apache/beam/sdk/io/elasticsearch/ElasticsearchIO.java ---------------------------------------------------------------------- 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 e3965dc..fa67fe1 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 @@ -455,16 +455,7 @@ public class ElasticsearchIO { while (shards.hasNext()) { Map.Entry<String, JsonNode> shardJson = shards.next(); String shardId = shardJson.getKey(); - JsonNode value = (JsonNode) shardJson.getValue(); - boolean isPrimaryShard = - value - .path(0) - .path("routing") - .path("primary") - .asBoolean(); - if (isPrimaryShard) { - sources.add(new BoundedElasticsearchSource(spec, shardId)); - } + sources.add(new BoundedElasticsearchSource(spec, shardId)); } checkArgument(!sources.isEmpty(), "No primary shard found"); return sources;
