[MR] Check for invalid Progressable

In Hadoop-like envs, Progressable might be invalid causing exceptions when going through the MR layer. Fix #330
elastic · Dec 10, 2014 · 945ccca · 945ccca
1 parent da47ce8
commit 945ccca
Show file tree

Hide file tree

Showing 2 changed files with 12 additions and 4 deletions.
diff --git a/mr/src/main/java/org/elasticsearch/hadoop/mr/EsInputFormat.java b/mr/src/main/java/org/elasticsearch/hadoop/mr/EsInputFormat.java
@@ -225,7 +225,10 @@ void init(ShardInputSplit esSplit, Configuration cfg, Progressable progressable)
             scrollReader = new ScrollReader(reader, mapping);
 
             // heart-beat
-            beat = new HeartBeat(progressable, cfg, settings.getHeartBeatLead(), log);
+            // in Hadoop-like envs (Spark) the progressable might be null and thus the heart-beat is not needed
+			if (progressable != null) {
+				beat = new HeartBeat(progressable, cfg, settings.getHeartBeatLead(), log);
+			}
 
             // initialize REST client
             client = new RestRepository(settings);
@@ -308,7 +311,9 @@ public void close() throws IOException {
         @Override
         public boolean next(K key, V value) throws IOException {
             if (scrollQuery == null) {
-                beat.start();
+				if (beat != null) {
+					beat.start();
+				}
 
                 scrollQuery = queryBuilder.build(client, scrollReader);
                 size = scrollQuery.getSize();

diff --git a/mr/src/main/java/org/elasticsearch/hadoop/mr/EsOutputFormat.java b/mr/src/main/java/org/elasticsearch/hadoop/mr/EsOutputFormat.java
@@ -184,8 +184,11 @@ protected void init() throws IOException {
             // select the appropriate nodes first, to spread the load before-hand
             SettingsUtils.pinNode(settings, nodes.get(currentInstance % nodes.size()));
 
-            beat = new HeartBeat(progressable, cfg, settings.getHeartBeatLead(), log);
-            beat.start();
+			// in Hadoop-like envs (Spark) the progressable might be null and thus the heart-beat is not needed
+			if (progressable != null) {
+				beat = new HeartBeat(progressable, cfg, settings.getHeartBeatLead(), log);
+				beat.start();
+			}
 
             resource = new Resource(settings, false);