From 748fb9c16947a388acc402e3a127ad7eb170afe2 Mon Sep 17 00:00:00 2001
From: Andrew Ash <andrew@andrewash.com>
Date: Mon, 12 Dec 2016 18:52:46 -0800
Subject: [PATCH] SPARK-18113 Authorize duplicate requests to commit

Commonly encountered because of dropped responses from the driver back to an
executor authorizing a commit.
---
 .../org/apache/spark/scheduler/OutputCommitCoordinator.scala | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/core/src/main/scala/org/apache/spark/scheduler/OutputCommitCoordinator.scala b/core/src/main/scala/org/apache/spark/scheduler/OutputCommitCoordinator.scala
index 7bed6851d0cde..4f1fee1663a61 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/OutputCommitCoordinator.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/OutputCommitCoordinator.scala
@@ -164,6 +164,11 @@ private[spark] class OutputCommitCoordinator(conf: SparkConf, isDriver: Boolean)
               s"partition=$partition")
             authorizedCommitters(partition) = attemptNumber
             true
+          case existingCommitter if existingCommitter == attemptNumber =>
+            logWarning(s"Authorizing duplicate request to commit for " +
+              s"attemptNumber=$attemptNumber to commit for stage=$stage, partition=$partition; " +
+              s"existingCommitter = $existingCommitter. This can indicate dropped network traffic.")
+            true
           case existingCommitter =>
             logDebug(s"Denying attemptNumber=$attemptNumber to commit for stage=$stage, " +
               s"partition=$partition; existingCommitter = $existingCommitter")