Merge pull request #65 from svenkreiss/failing-partitions

prevent failing partition exceptions from bubbling up
svenkreiss · Jun 12, 2017 · a9463c5 · a9463c5
2 parents 51766c7 + 4e93538
commit a9463c5
Showing 1 changed file with 9 additions and 8 deletions.
diff --git a/pysparkling/context.py b/pysparkling/context.py
@@ -49,14 +49,15 @@ def _run_task(task_context, rdd, func, partition):
                  ''.format(task_context.attempt_number, partition.index,
                            rdd.name()))
         traceback.print_exc()
-        if task_context.attempt_number == task_context.max_retries:
-            log.error('Partition {} of {} failed.'
-                      ''.format(partition.index, rdd.name()))
-            return []
-
-        if task_context.retry_wait:
-            time.sleep(task_context.retry_wait)
-        return _run_task(task_context, rdd, func, partition)
+
+    if task_context.attempt_number == task_context.max_retries:
+        log.error('Partition {} of {} failed.'
+                  ''.format(partition.index, rdd.name()))
+        return []
+
+    if task_context.retry_wait:
+        time.sleep(task_context.retry_wait)
+    return _run_task(task_context, rdd, func, partition)
 
 
 def runJob_map(i):