[SPARK-10311][Streaming]Reload appId and attemptId when app starts with checkpoint file in cluster mode

[XuTingjun]

No description provided.

[SparkQA]

Test build #41670 has finished for PR 8477 at commit 3211a68.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[tdas]

@harishreedharan @vanzin Could you guys take a look at this?

[vanzin]

This doesn't look right. If the AM restarted, the attempt ID will be different.

[vanzin]

I'm not familiar with how this data is used by the streaming backend, but it looks odd to be manually reloading it from a checkpoint.

In client mode, AM restarts do not affect the driver. In cluster mode, the app id is the same, but the attempt id changes; and those will be re-populated anyway when the SparkContext for the app is initialized by the new AM process.

[XuTingjun]

Sorry that I don't declare the problem clearly.

When an app starts with CheckPoint file using getOrCreate method, the new AM process will new a SparkContext object, but just using the old SparkConf, So the new attemptId set by new AM process doesn't do anything.

 val newSparkConf = new SparkConf(loadDefaults = false).setAll(sparkConfPairs)

Also the appId is the same.

[harishreedharan]

+1. This looks good.

@vanzin - I assume your concern is with the reloading. Reloading here is to ignore the old values from the serlialized spark conf so we pick up the values in the new one which was created by the current attempt.

[vanzin]

@tdas Hari knows a lot more about this area than me so feel free to ignore my comments.

[tdas]

@harishreedharan Please take a look.
@XuTingjun Could you update the Description of this PR (the area below the title) with the explanation you gave to @vanzin ??

[harishreedharan]

I already +1-ed it. LGTM

[tdas]

@harishreedharan Oops, yeah, missed that. Merging this to master and 1.5