[SPARK-4346][SPARK-3596][YARN] Commonize the monitor logic

[Sephiroth-Lin]

1. YarnClientSchedulerBack.asyncMonitorApplication use Client.monitorApplication so that commonize the monitor logic
2. Support changing the yarn client monitor interval, see [SPARK-3596][YARN]Support changing the yarn client monitor interval #5292
3. More details see discussion on [SPARK-4282][YARN] Stopping flag in YarnClientSchedulerBackend should be volatile #3143

[srowen]

Jenkins, add to whitelist

[srowen]

ok to test

[SparkQA]

Test build #29536 has finished for PR 5305 at commit 6b47ff7.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.
• This patch does not change any dependencies.

[Sephiroth-Lin]

Jenkins, retest please

[Sephiroth-Lin]

@srowen unit tests failed at run Python app on yarn-cluster mode, I think this didn't cause by this PR, please ask jenkins to retest, thank you.

[srowen]

Jenkins, retest this please.

[SparkQA]

Test build #29604 has finished for PR 5305 at commit 6b47ff7.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.
• This patch does not change any dependencies.

[srowen]

CC a few people who have touched this bit of the code: @kasjain @witgo @andrewor14

[witgo]

The process doesn't look the same. ApplicationNotFoundException is not handled.

[andrewor14]

Yes, we can just add it in Client#monitorApplication as well

[andrewor14]

Hi @Sephiroth-Lin this looks good for the most part. The only reason why I didn't merge these code paths initially is because the loop here checks on stopping, and I didn't want to pass the boolean into monitorApplication. However, on second examination this is a daemon thread anyway, so it will eventually exit regardless of whether the backend has been stopped.

Once you address @witgo's comment I will merge this thanks.

[SparkQA]

Test build #29643 has finished for PR 5305 at commit 6483a2a.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.
• This patch does not change any dependencies.

[SparkQA]

Test build #29644 has finished for PR 5305 at commit ee2b2fd.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.
• This patch does not change any dependencies.

[andrewor14]

you probably don't need to wrap this entire block in the try. You could for instance do the following instead:

val report: ApplicationReport =
  try {
    getApplicationReport(appId)
  } catch {
    case e: ApplicationNotFoundException =>
      return (..., ...)
  }
val state = report.getYarnApplicationState

[Sephiroth-Lin]

Done. Thank you!!!

[tgravescs]

stop() can be called from other areas of the code (like SparkContext.stop()). Now that the loop isn't checking for it we wouldn't interrupt this thread if that happens and I think we need to handle that case. See the discussions on #3143

[andrewor14]

good point. We might need to put this in a private var monitorThread or something and interrupt it in stop(). (@Sephiroth-Lin I would make this method return the thread and set monitorThread = asyncMonitorApplication() in start())

[Sephiroth-Lin]

Right. We need to interrupt the thread in stop().

[SparkQA]

Test build #29827 has finished for PR 5305 at commit 47c0014.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.
• This patch does not change any dependencies.

[andrewor14]

Nice, LGTM merging into master thanks @Sephiroth-Lin @tgravescs @witgo