[SPARK-37147] MetricsReporter producing NullPointerException when element 'triggerExecution' not present in Map[]

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 3.2.0
• Fix Version/s: 3.2.1, 3.3.0
• Component/s: Structured Streaming

Language

scala

• Assignee: Radoslaw Busz
• Reporter: Radoslaw Busz
• Created: 28/Oct/21 16:27
• Updated: 30/Oct/21 22:48
• Resolved: 30/Oct/21 22:48

Description

The exception occurs in MetricsReporter when it tries to register gauges using lastProgress of each stream. This problem was partially fixed in https://issues.apache.org/jira/browse/SPARK-22975 but in introduced a NPE when 'triggerExecution' element is not present in durationMS map:

 

registerGauge("latency", _.durationMs.get("triggerExecution").longValue(), 0L)

 

I find it difficult to reproduce this every time but it happens every few restarts when structured streaming uses a slow event source (very little or no events). In my case it breaks metric reporting via Codehale/Dropwizard and generates multiple stacktraces such as:

21/09/16 09:51:36 ERROR ScheduledReporter: Exception thrown from GangliaReporter#report. Exception was suppressed.
java.lang.NullPointerException
	at org.apache.spark.sql.execution.streaming.MetricsReporter.$anonfun$new$3(MetricsReporter.scala:43)
	at org.apache.spark.sql.execution.streaming.MetricsReporter.$anonfun$new$3$adapted(MetricsReporter.scala:43)
	at scala.Option.map(Option.scala:230)
	at org.apache.spark.sql.execution.streaming.MetricsReporter$$anon$1.getValue(MetricsReporter.scala:68)
	at com.codahale.metrics.ganglia.GangliaReporter.reportGauge(GangliaReporter.java:353)
	at com.codahale.metrics.ganglia.GangliaReporter.report(GangliaReporter.java:240)
	at com.codahale.metrics.ScheduledReporter.report(ScheduledReporter.java:237)
	at com.codahale.metrics.ScheduledReporter.lambda$start$0(ScheduledReporter.java:177)
	at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
	at java.util.concurrent.FutureTask.runAndReset(FutureTask.java:308)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$301(ScheduledThreadPoolExecutor.java:180)
	at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:294)
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
	at java.lang.Thread.run(Thread.java:748)

I'm happy to implement a fix.

Issue Links

• links to: [Github] Pull Request #34426 (gitplaneta)