[SPARK-3150] Fix NullPointerException in in Spark recovery: Add initializing default values in DriverInfo.init()

The issue happens when Spark is run standalone on a cluster.
When master and driver fall simultaneously on one node in a cluster, master tries to recover its state and restart spark driver.
While restarting driver, it falls with NPE exception (stacktrace is below).
After falling, it restarts and tries to recover its state and restart Spark driver again. It happens over and over in an infinite cycle.
Namely, Spark tries to read DriverInfo state from zookeeper, but after reading it happens to be null in DriverInfo.worker.

https://issues.apache.org/jira/browse/SPARK-3150

Author: Tatiana Borisova <tanyatik@yandex.ru>

Closes #2062 from tanyatik/spark-3150 and squashes the following commits:

9936043 [Tatiana Borisova] Add initializing default values in DriverInfo.init()

(cherry picked from commit 70d814665baa8b8ca868d3126452105ecfa5cbff)
Signed-off-by: Josh Rosen <joshrosen@apache.org>
[SPARK-3190] Avoid overflow in VertexRDD.count()

VertexRDDs with more than 4 billion elements are counted incorrectly due to integer overflow when summing partition sizes. This PR fixes the issue by converting partition sizes to Longs before summing them.

The following code previously returned -10000000. After applying this PR, it returns the correct answer of 5000000000 (5 billion).

```scala
val pairs = sc.parallelize(0L until 500L).map(_ * 10000000)
  .flatMap(start => start until (start + 10000000)).map(x => (x, x))
VertexRDD(pairs).count()
```

Author: Ankur Dave <ankurdave@gmail.com>

Closes #2106 from ankurdave/SPARK-3190 and squashes the following commits:

641f468 [Ankur Dave] Avoid overflow in VertexRDD.count()

(cherry picked from commit 96df92906978c5f58e0cc8ff5eebe5b35a08be3b)
Signed-off-by: Josh Rosen <joshrosen@apache.org>
[SPARK-2981][GraphX] EdgePartition1D Int overflow

minor fix
detail is here: https://issues.apache.org/jira/browse/SPARK-2981

Author: Larry Xiao <xiaodi@sjtu.edu.cn>

Closes #1902 from larryxiao/2981 and squashes the following commits:

88059a2 [Larry Xiao] [SPARK-2981][GraphX] EdgePartition1D Int overflow

(cherry picked from commit aa7de128c5987fd2e134736f07ae913ad1f5eb26)
Signed-off-by: Ankur Dave <ankurdave@gmail.com>
[SPARK-2823][GraphX]fix GraphX EdgeRDD zipPartitions

If the users set “spark.default.parallelism” and the value is different with the EdgeRDD partition number, GraphX jobs will throw:
java.lang.IllegalArgumentException: Can't zip RDDs with unequal numbers of partitions

Author: luluorta <luluorta@gmail.com>

Closes #1763 from luluorta/fix-graph-zip and squashes the following commits:

8338961 [luluorta] fix GraphX EdgeRDD zipPartitions

(cherry picked from commit 9b225ac3072de522b40b46aba6df1f1c231f13ef)
Signed-off-by: Ankur Dave <ankurdave@gmail.com>
[SPARK-3216] [SPARK-3232] Spark-shell is broken in branch-1.0 / Backport SPARK-3006 into branch-1.0

Author: Kousuke Saruta <sarutak@oss.nttdata.co.jp>
Author: Andrew Or <andrewor14@gmail.com>

Closes #2136 from sarutak/SPARK-3216 and squashes the following commits:

5c20d2d [Kousuke Saruta] Merge branch 'branch-1.0' of git://git.apache.org/spark into SPARK-3216
af0517f [Kousuke Saruta] Modified spark-shell.cmd, backporting SPARK-3006
bbc7221 [Kousuke Saruta] Replace 4-space into 2-space
15cd980 [Andrew Or] Fix spark-shell in branch-1.0
[HOTFIX] [SPARK-3400] Revert 9b225ac "fix GraphX EdgeRDD zipPartitions"

9b225ac3072de522b40b46aba6df1f1c231f13ef has been causing GraphX tests
to fail nondeterministically, which is blocking development for others.

Author: Ankur Dave <ankurdave@gmail.com>

Closes #2271 from ankurdave/SPARK-3400 and squashes the following commits:

10c2a97 [Ankur Dave] [HOTFIX] [SPARK-3400] Revert 9b225ac "fix GraphX EdgeRDD zipPartitions"

(cherry picked from commit 00362dac976cd05b06638deb11d990d612429e0b)
Signed-off-by: Ankur Dave <ankurdave@gmail.com>
[SPARK-3394] [SQL] Fix crash in TakeOrdered when limit is 0

This resolves https://issues.apache.org/jira/browse/SPARK-3394

Author: Eric Liang <ekl@google.com>

Closes #2264 from ericl/spark-3394 and squashes the following commits:

c87355b [Eric Liang] refactor
bfb6140 [Eric Liang] change RDD takeOrdered instead
7a51528 [Eric Liang] fix takeordered when limit = 0

(cherry picked from commit 6754570d83044c4fbaf0d2ac2378a0e081a93629)
Signed-off-by: Matei Zaharia <matei@databricks.com>
[SPARK-3481] [SQL] Eliminate the error log in local Hive comparison test

Logically, we should remove the Hive Table/Database first and then reset the Hive configuration, repoint to the new data warehouse directory etc.
Otherwise it raised exceptions like "Database doesn't not exists: default" in the local testing.

Author: Cheng Hao <hao.cheng@intel.com>

Closes #2352 from chenghao-intel/test_hive and squashes the following commits:

74fd76b [Cheng Hao] eliminate the error log

(cherry picked from commit 8194fc662c08eb445444c207264e22361def54ea)
Signed-off-by: Michael Armbrust <michael@databricks.com>
[SPARK-3709] Executors don't always report broadcast block removal properly back to the driver (for branch-1.1)

Author: Reynold Xin <rxin@apache.org>

Closes #2591 from rxin/SPARK-3709-1.1 and squashes the following commits:

ab99cc0 [Reynold Xin] [SPARK-3709] Executors don't always report broadcast block removal properly back to the driver

(cherry picked from commit a8c6e82de31cb09b7e7e37caf4e52c9e10185b63)
Signed-off-by: Reynold Xin <rxin@apache.org>
[SPARK-3121] Wrong implementation of implicit bytesWritableConverter

val path = ... //path to seq file with BytesWritable as type of both key and value
val file = sc.sequenceFile[Array[Byte],Array[Byte]](path)
file.take(1)(0)._1

This prints incorrect content of byte array. Actual content starts with correct one and some "random" bytes and zeros are appended. BytesWritable has two methods:

getBytes() - return content of all internal array which is often longer then actual value stored. It usually contains the rest of previous longer values

copyBytes() - return just begining of internal array determined by internal length property

It looks like in implicit conversion between BytesWritable and Array[byte] getBytes is used instead of correct copyBytes.

dbtsai

Author: Jakub Dubovský <james64@inMail.sk>
Author: Dubovsky Jakub <dubovsky@avast.com>

Closes #2712 from james64/3121-bugfix and squashes the following commits:

f85d24c [Jakub Dubovský] Test name changed, comments added
1b20d51 [Jakub Dubovský] Import placed correctly
406e26c [Jakub Dubovský] Scala style fixed
f92ffa6 [Dubovsky Jakub] performance tuning
480f9cd [Dubovsky Jakub] Bug 3121 fixed

(cherry picked from commit fc616d51a510f82627b5be949a5941419834cf70)
Signed-off-by: Josh Rosen <joshrosen@apache.org>
[SPARK-2546] Clone JobConf for each task (branch-1.0 / 1.1 backport)

This patch attempts to fix SPARK-2546 in `branch-1.0` and `branch-1.1`.  The underlying problem is that thread-safety issues in Hadoop Configuration objects may cause Spark tasks to get stuck in infinite loops.  The approach taken here is to clone a new copy of the JobConf for each task rather than sharing a single copy between tasks.  Note that there are still Configuration thread-safety issues that may affect the driver, but these seem much less likely to occur in practice and will be more complex to fix (see discussion on the SPARK-2546 ticket).

This cloning is guarded by a new configuration option (`spark.hadoop.cloneConf`) and is disabled by default in order to avoid unexpected performance regressions for workloads that are unaffected by the Configuration thread-safety issues.

Author: Josh Rosen <joshrosen@apache.org>

Closes #2684 from JoshRosen/jobconf-fix-backport and squashes the following commits:

f14f259 [Josh Rosen] Add configuration option to control cloning of Hadoop JobConf.
b562451 [Josh Rosen] Remove unused jobConfCacheKey field.
dd25697 [Josh Rosen] [SPARK-2546] [1.0 / 1.1 backport] Clone JobConf for each task.

(cherry picked from commit 2cd40db2b3ab5ddcb323fd05c171dbd9025f9e71)
Signed-off-by: Josh Rosen <joshrosen@databricks.com>

Conflicts:
	docs/configuration.md
[SPARK-4109][CORE] Correctly deserialize Task.stageId

The two subclasses of Task, ShuffleMapTask and ResultTask, do not correctly deserialize stageId. Therefore, the accessing of TaskContext.stageId always returns zero value to the user.

Author: luluorta <luluorta@gmail.com>

Closes #2971 from luluorta/fix-task-ser and squashes the following commits:

ff35ee6 [luluorta] correctly deserialize Task.stageId
[SPARK-4249][GraphX]fix a problem of EdgePartitionBuilder in Graphx

at first srcIds is not initialized and are all 0. so we use edgeArray(0).srcId to currSrcId

Author: lianhuiwang <lianhuiwang09@gmail.com>

Closes #3138 from lianhuiwang/SPARK-4249 and squashes the following commits:

3f4e503 [lianhuiwang] fix a problem of EdgePartitionBuilder in Graphx

(cherry picked from commit d15c6e9dc2860bbe56e31ddf71218ccc6d5c841d)
Signed-off-by: Ankur Dave <ankurdave@gmail.com>
Update JavaCustomReceiver.java

数组下标越界

Author: xiao321 <1042460381@qq.com>

Closes #3153 from xiao321/patch-1 and squashes the following commits:

0ed17b5 [xiao321] Update JavaCustomReceiver.java

(cherry picked from commit 7c9ec529a3483fab48f728481dd1d3663369e50a)
Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>
[SPARK-4304] [PySpark] Fix sort on empty RDD (1.0 branch)

This PR fix sortBy()/sortByKey() on empty RDD.

This should be back ported into 1.0

Author: Davies Liu <davies@databricks.com>

Closes #3163 from davies/fix_sort_1.0 and squashes the following commits:

9be984f [Davies Liu] fix sort on empty RDD
[SPARK-4301] StreamingContext should not allow start() to be called after calling stop()

In Spark 1.0.0+, calling `stop()` on a StreamingContext that has not been started is a no-op which has no side-effects. This allows users to call `stop()` on a fresh StreamingContext followed by `start()`. I believe that this almost always indicates an error and is not behavior that we should support. Since we don't allow `start() stop() start()` then I don't think it makes sense to allow `stop() start()`.

The current behavior can lead to resource leaks when StreamingContext constructs its own SparkContext: if I call `stop(stopSparkContext=True)`, then I expect StreamingContext's underlying SparkContext to be stopped irrespective of whether the StreamingContext has been started. This is useful when writing unit test fixtures.

Prior discussions:
- https://github.com/apache/spark/pull/3053#discussion-diff-19710333R490
- https://github.com/apache/spark/pull/3121#issuecomment-61927353

Author: Josh Rosen <joshrosen@databricks.com>

Closes #3160 from JoshRosen/SPARK-4301 and squashes the following commits:

dbcc929 [Josh Rosen] Address more review comments
bdbe5da [Josh Rosen] Stop SparkContext after stopping scheduler, not before.
03e9c40 [Josh Rosen] Always stop SparkContext, even if stop(false) has already been called.
832a7f4 [Josh Rosen] Address review comment
5142517 [Josh Rosen] Add tests; improve Scaladoc.
813e471 [Josh Rosen] Revert workaround added in https://github.com/apache/spark/pull/3053/files#diff-e144dbee130ed84f9465853ddce65f8eR49
5558e70 [Josh Rosen] StreamingContext.stop() should stop SparkContext even if StreamingContext has not been started yet.

(cherry picked from commit 7b41b17f3296eea3282efbdceb6b28baf128287d)
Signed-off-by: Tathagata Das <tathagata.das1565@gmail.com>
Update RecoverableNetworkWordCount.scala

Trying this example, I missed the moment when the checkpoint was iniciated

Author: comcmipi <pitonak@fns.uniba.sk>

Closes #2735 from comcmipi/patch-1 and squashes the following commits:

b6d8001 [comcmipi] Update RecoverableNetworkWordCount.scala
96fe274 [comcmipi] Update RecoverableNetworkWordCount.scala
[SPARK-4433] fix a racing condition in zipWithIndex

Spark hangs with the following code:

~~~
sc.parallelize(1 to 10).zipWithIndex.repartition(10).count()
~~~

This is because ZippedWithIndexRDD triggers a job in getPartitions and it causes a deadlock in DAGScheduler.getPreferredLocs (synced). The fix is to compute `startIndices` during construction.

This should be applied to branch-1.0, branch-1.1, and branch-1.2.

pwendell

Author: Xiangrui Meng <meng@databricks.com>

Closes #3291 from mengxr/SPARK-4433 and squashes the following commits:

c284d9f [Xiangrui Meng] fix a racing condition in zipWithIndex

(cherry picked from commit bb46046154a438df4db30a0e1fd557bd3399ee7b)
Signed-off-by: Xiangrui Meng <meng@databricks.com>
[SPARK-4597] Use proper exception and reset variable in Utils.createTempDir()

`File.exists()` and `File.mkdirs()` only throw `SecurityException` instead of `IOException`. Then, when an exception is thrown, `dir` should be reset too.

Author: Liang-Chi Hsieh <viirya@gmail.com>

Closes #3449 from viirya/fix_createtempdir and squashes the following commits:

36cacbd [Liang-Chi Hsieh] Use proper exception and reset variable.

(cherry picked from commit 49fe8797e64f10c574e0790b32a8c3fdc7e594a0)
Signed-off-by: Josh Rosen <joshrosen@databricks.com>