[BUG] Spark Jobs running indefinitely in some VMs, causing job failure

@svotaw is already working on this.

Since Apr 18, 2023.

### SynapseML version

0.9.5

### System information

- **Language version** (e.g. python 3.8, scala 2.12): Scala 2.12
- **Spark Version** (e.g. 3.2.3): Spark 3.2.1
- **Spark Platform** (e.g. Synapse, Databricks): Databricks

### Describe the problem

I have a MS customer with an issue and it seems to be related to the library.  We have engaged Databricks PG, and they find some messages related to tasks not making progress as below:

 

23/03/12 03:41:10 WARN HangingTaskDetector: Doing a full thread dump to debug potential hanging tasks.
23/03/12 03:41:10 INFO privateLog: "Executor task launch worker for task 25.0 in stage 500.0 (TID 43006)" #1210 WAITING holding [Lock(java.util.concurrent.ThreadPoolExecutor$Worker@1499025289})]
sun.misc.Unsafe.park(Native Method)
java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:837)
java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:999)
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1308)
java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
com.microsoft.azure.synapse.ml.lightgbm.LightGBMBase.trainLightGBM(LightGBMBase.scala:367)
com.microsoft.azure.synapse.ml.lightgbm.LightGBMBase.$anonfun$innerTrain$4(LightGBMBase.scala:485)
com.microsoft.azure.synapse.ml.lightgbm.LightGBMBase$$Lambda$4994/514322416.apply(Unknown Source)
org.apache.spark.sql.execution.MapPartitionsExec.$anonfun$doExecute$3(objects.scala:228)
org.apache.spark.sql.execution.MapPartitionsExec$$Lambda$4995/1261819360.apply(Unknown Source)
org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:903)
org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:903)
org.apache.spark.rdd.RDD$$Lambda$4089/175029406.apply(Unknown Source)
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
org.apache.spark.sql.execution.SQLExecutionRDD.compute(SQLExecutionRDD.scala:60)
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
org.apache.spark.scheduler.ResultTask.$anonfun$runTask$3(ResultTask.scala:75)
org.apache.spark.scheduler.ResultTask$$Lambda$4770/1953552051.apply(Unknown Source)
com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
org.apache.spark.scheduler.ResultTask.$anonfun$runTask$1(ResultTask.scala:75)
org.apache.spark.scheduler.ResultTask$$Lambda$4752/1067568607.apply(Unknown Source)
com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:55)
org.apache.spark.scheduler.Task.doRunTask(Task.scala:156)
org.apache.spark.scheduler.Task.$anonfun$run$1(Task.scala:125)
org.apache.spark.scheduler.Task$$Lambda$1165/1147243086.apply(Unknown Source)
com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
org.apache.spark.scheduler.Task.run(Task.scala:95)
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$13(Executor.scala:832)
org.apache.spark.executor.Executor$TaskRunner$$Lambda$1127/346961738.apply(Unknown Source)
org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1681)
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:835)
org.apache.spark.executor.Executor$TaskRunner$$Lambda$912/2007430619.apply$mcV$sp(Unknown Source)
scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:690)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)

This is happening intermittently and in a few VMs causing the job to get running indefinitely without any progress until the customer ends it. I will ask the customer in parallel to open an issue in Github to check on it.

We could not identify a high use of resources - VMs - that could explain the issue with some of the VMs. The data being manipulated is about 50mb, not so big.

I would like your kind guidance how to get help on this as fast as possible as the customer is really impacted.

### Code to reproduce issue

More information can be shared.

### Other info / logs

23/03/12 03:41:10 WARN HangingTaskDetector: Task 42986 is probably not making progress because its metrics (Map(internal.metrics.shuffle.read.localBlocksFetched -> 0, internal.metrics.shuffle.read.remoteBytesReadToDisk -> 0, internal.metrics.shuffle.write.bytesWritten -> 0, internal.metrics.output.recordsWritten -> 0, internal.metrics.shuffle.write.recordsWritten -> 0, internal.metrics.memoryBytesSpilled -> 0, internal.metrics.shuffle.read.remoteBytesRead -> 0, internal.metrics.diskBytesSpilled -> 0, internal.metrics.shuffle.read.localBytesRead -> 0, internal.metrics.shuffle.read.recordsRead -> 0, internal.metrics.output.bytesWritten -> 0, internal.metrics.input.bytesRead -> 26206121, internal.metrics.input.recordsRead -> 4096, internal.metrics.shuffle.read.remoteBlocksFetched -> 0)) has not changed since Sun Mar 12 03:31:00 UTC 2023
23/03/12 03:41:10 WARN HangingTaskDetector: Doing a full thread dump to debug potential hanging tasks.
23/03/12 03:41:10 INFO privateLog: "Executor task launch worker for task 25.0 in stage 500.0 (TID 43006)" #1210 WAITING holding [Lock(java.util.concurrent.ThreadPoolExecutor$Worker@1499025289})]
sun.misc.Unsafe.park(Native Method)
java.util.concurrent.locks.LockSupport.park(LockSupport.java:175)
java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:837)
java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:999)
java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1308)
java.util.concurrent.CountDownLatch.await(CountDownLatch.java:231)
com.microsoft.azure.synapse.ml.lightgbm.LightGBMBase.trainLightGBM(LightGBMBase.scala:367)
com.microsoft.azure.synapse.ml.lightgbm.LightGBMBase.$anonfun$innerTrain$4(LightGBMBase.scala:485)
com.microsoft.azure.synapse.ml.lightgbm.LightGBMBase$$Lambda$4994/514322416.apply(Unknown Source)
org.apache.spark.sql.execution.MapPartitionsExec.$anonfun$doExecute$3(objects.scala:228)
org.apache.spark.sql.execution.MapPartitionsExec$$Lambda$4995/1261819360.apply(Unknown Source)
org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2(RDD.scala:903)
org.apache.spark.rdd.RDD.$anonfun$mapPartitionsInternal$2$adapted(RDD.scala:903)
org.apache.spark.rdd.RDD$$Lambda$4089/175029406.apply(Unknown Source)
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
org.apache.spark.sql.execution.SQLExecutionRDD.compute(SQLExecutionRDD.scala:60)
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:60)
org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:380)
org.apache.spark.rdd.RDD.iterator(RDD.scala:344)
org.apache.spark.scheduler.ResultTask.$anonfun$runTask$3(ResultTask.scala:75)
org.apache.spark.scheduler.ResultTask$$Lambda$4770/1953552051.apply(Unknown Source)
com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
org.apache.spark.scheduler.ResultTask.$anonfun$runTask$1(ResultTask.scala:75)
org.apache.spark.scheduler.ResultTask$$Lambda$4752/1067568607.apply(Unknown Source)
com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:55)
org.apache.spark.scheduler.Task.doRunTask(Task.scala:156)
org.apache.spark.scheduler.Task.$anonfun$run$1(Task.scala:125)
org.apache.spark.scheduler.Task$$Lambda$1165/1147243086.apply(Unknown Source)
com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
org.apache.spark.scheduler.Task.run(Task.scala:95)
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$13(Executor.scala:832)
org.apache.spark.executor.Executor$TaskRunner$$Lambda$1127/346961738.apply(Unknown Source)
org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1681)
org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$4(Executor.scala:835)
org.apache.spark.executor.Executor$TaskRunner$$Lambda$912/2007430619.apply$mcV$sp(Unknown Source)
scala.runtime.java8.JFunction0$mcV$sp.apply(JFunction0$mcV$sp.java:23)
com.databricks.spark.util.ExecutorFrameProfiler$.record(ExecutorFrameProfiler.scala:110)
org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:690)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
java.lang.Thread.run(Thread.java:750)
.....
.....
.....
23/03/12 03:41:10 WARN HangingTaskDetector: Task 43006 is probably not making progress because its metrics (Map(internal.metrics.shuffle.read.localBlocksFetched -> 0, internal.metrics.shuffle.read.remoteBytesReadToDisk -> 0, internal.metrics.shuffle.write.bytesWritten -> 0, internal.metrics.output.recordsWritten -> 0, internal.metrics.shuffle.write.recordsWritten -> 0, internal.metrics.memoryBytesSpilled -> 0, internal.metrics.shuffle.read.remoteBytesRead -> 0, internal.metrics.diskBytesSpilled -> 0, internal.metrics.shuffle.read.localBytesRead -> 0, internal.metrics.shuffle.read.recordsRead -> 0, internal.metrics.output.bytesWritten -> 0, internal.metrics.input.bytesRead -> 25829357, internal.metrics.input.recordsRead -> 4096, internal.metrics.shuffle.read.remoteBlocksFetched -> 0)) has not changed since Sun Mar 12 03:31:00 UTC 2023
23/03/12 03:41:20 WARN HangingTaskDetector: Task 42986 is probably not making progress because its metrics (Map(internal.metrics.shuffle.read.localBlocksFetched -> 0, internal.metrics.shuffle.read.remoteBytesReadToDisk -> 0, internal.metrics.shuffle.write.bytesWritten -> 0, internal.metrics.output.recordsWritten -> 0, internal.metrics.shuffle.write.recordsWritten -> 0, internal.metrics.memoryBytesSpilled -> 0, internal.metrics.shuffle.read.remoteBytesRead -> 0, internal.metrics.diskBytesSpilled -> 0, internal.metrics.shuffle.read.localBytesRead -> 0, internal.metrics.shuffle.read.recordsRead -> 0, internal.metrics.output.bytesWritten -> 0, internal.metrics.input.bytesRead -> 26206121, internal.metrics.input.recordsRead -> 4096, internal.metrics.shuffle.read.remoteBlocksFetched -> 0)) has not changed since Sun Mar 12 03:31:00 UTC 2023
23/03/12 03:41:20 WARN HangingTaskDetector: Task 43006 is probably not making progress because its metrics (Map(internal.metrics.shuffle.read.localBlocksFetched -> 0, internal.metrics.shuffle.read.remoteBytesReadToDisk -> 0, internal.metrics.shuffle.write.bytesWritten -> 0, internal.metrics.output.recordsWritten -> 0, internal.metrics.shuffle.write.recordsWritten -> 0, internal.metrics.memoryBytesSpilled -> 0, internal.metrics.shuffle.read.remoteBytesRead -> 0, internal.metrics.diskBytesSpilled -> 0, internal.metrics.shuffle.read.localBytesRead -> 0, internal.metrics.shuffle.read.recordsRead -> 0, internal.metrics.output.bytesWritten -> 0, internal.metrics.input.bytesRead -> 25829357, internal.metrics.input.recordsRead -> 4096, internal.metrics.shuffle.read.remoteBlocksFetched -> 0)) has not changed since Sun Mar 12 03:31:00 UTC 2023
23/03/12 03:41:30 WARN HangingTaskDetector: Task 42986 is probably not making progress because its metrics (Map(internal.metrics.shuffle.read.localBlocksFetched -> 0, internal.metrics.shuffle.read.remoteBytesReadToDisk -> 0, internal.metrics.shuffle.write.bytesWritten -> 0, internal.metrics.output.recordsWritten -> 0, internal.metrics.shuffle.write.recordsWritten -> 0, internal.metrics.memoryBytesSpilled -> 0, internal.metrics.shuffle.read.remoteBytesRead -> 0, internal.metrics.diskBytesSpilled -> 0, internal.metrics.shuffle.read.localBytesRead -> 0, internal.metrics.shuffle.read.recordsRead -> 0, internal.metrics.output.bytesWritten -> 0, internal.metrics.input.bytesRead -> 26206121, internal.metrics.input.recordsRead -> 4096, internal.metrics.shuffle.read.remoteBlocksFetched -> 0)) has not changed since Sun Mar 12 03:31:00 UTC 2023

### What component(s) does this bug affect?

- [ ] `area/cognitive`: Cognitive project
- [ ] `area/core`: Core project
- [ ] `area/deep-learning`: DeepLearning project
- [X] `area/lightgbm`: Lightgbm project
- [ ] `area/opencv`: Opencv project
- [ ] `area/vw`: VW project
- [ ] `area/website`: Website
- [ ] `area/build`: Project build system
- [ ] `area/notebooks`: Samples under notebooks folder
- [ ] `area/docker`: Docker usage
- [ ] `area/models`: models related issue

### What language(s) does this bug affect?

- [X] `language/scala`: Scala source code
- [ ] `language/python`: Pyspark APIs
- [ ] `language/r`: R APIs
- [ ] `language/csharp`: .NET APIs
- [ ] `language/new`: Proposals for new client languages

### What integration(s) does this bug affect?

- [ ] `integrations/synapse`: Azure Synapse integrations
- [ ] `integrations/azureml`: Azure ML integrations
- [X] `integrations/databricks`: Databricks integrations