[SPARK-19438] Both reading and updating executorDataMap should be guarded by CoarseGrainedSchedulerBackend.this.synchronized when handle RegisterExecutor.

[jinxing64]

What changes were proposed in this pull request?

Currently when handle RegisterExecutor in CoarseGrainedSchedulerBackend, executorDataMap is guarded by CoarseGrainedSchedulerBackend.this.synchronized when updating, which can cause numPendingExecutors incorrect.
Code is like below:

        if (executorDataMap.contains(executorId)) {
          executorRef.send(RegisterExecutorFailed("Duplicate executor ID: " + executorId))
          context.reply(true)
        } else {
          ...
          CoarseGrainedSchedulerBackend.this.synchronized {
            executorDataMap.put(executorId, data)
            if (currentExecutorIdCounter < executorId.toInt) {
              currentExecutorIdCounter = executorId.toInt
            }
            if (numPendingExecutors > 0) {
              numPendingExecutors -= 1
              logDebug(s"Decremented number of pending executors ($numPendingExecutors left)")
            }
          }

Consider SPARK-19437 and a scenario like below:
An executor sent RegisterExecutor twice by askWithRetry, and the interval between the two is quite small. Thus it might be possible that both of them will go to else branch, thus numPendingExecutors will be deducted twice and become incorrect. Currently, the askWithRetry of RegisterExecutor only exists in some unit tests, but it makes sense to make it stronger when handling RegisterExecutor.

TO FIX
Use CoarseGrainedSchedulerBackend.this.synchronized to guard executorDataMap when both reading and updating.

How was this patch tested?

The code logic is not changed, just manually tested. SPARK-19437 can also help verify this fix.

[AmplabJenkins]

Can one of the admins verify this patch?

[zsxwing]

Thus it might be possible that both of them will go to else branch

receiveAnyReply processes messages in sequence just like using one thread. This won't happen.

[jinxing64]

Thanks a lot for looking into this~ @zsxwing
You are right. My understanding about this is incorrect. CoarseGrainedSchedulerBackend: DriverEndpoint is a ThreadSafeRpcEndpoint, thus concurrent message processing is disabled.
I'll close this PR.