[SPARK-52346][SDP] Fix counter initialization and increment logic for flow retries

[JiaqiWang18]

What changes were proposed in this pull request?

Fixes flow failure counter increments by switching to updateWith. This prevents first increment from throwing NoSuchElementException when key does not exist.

Why are the changes needed?

Current impl flowToNumConsecutiveFailure(flowIdentifier) throws exception when a flow is retried for first time:

java.util.NoSuchElementException: key not found: `spark_catalog`.`test_db`.`mv`
	at scala.collection.MapOps.default(Map.scala:289)
	at scala.collection.MapOps.default$(Map.scala:288)
	at scala.collection.AbstractMap.default(Map.scala:420)
	at scala.collection.MapOps.apply(Map.scala:176)
	at scala.collection.MapOps.apply$(Map.scala:175)
	at scala.collection.AbstractMap.apply(Map.scala:420)
	at org.apache.spark.sql.pipelines.graph.GraphExecution.incrementFlowToNumConsecutiveFailure(GraphExecution.scala:52)
	at org.apache.spark.sql.pipelines.graph.GraphExecution.$anonfun$planAndStartFlow$1(GraphExecution.scala:92)
	at org.apache.spark.sql.pipelines.graph.GraphExecution.$anonfun$planAndStartFlow$1$adapted(GraphExecution.scala:90)
	at scala.concurrent.impl.Promise$Transformation.run(Promise.scala:484)
	at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)
	at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)
	at java.base/java.lang.Thread.run(Thread.java:840)

Does this PR introduce any user-facing change?

No

How was this patch tested?

Run pipeline manually with flow failure to test.

Was this patch authored or co-authored using generative AI tooling?

No

[JiaqiWang18]

@SCHJonathan @gengliangwang for review, thanks!

[JiaqiWang18]

Also @anishm-db

[itskals]

Looks good to me.

[gengliangwang]

@JiaqiWang18 shall we have a unit test for this one?

[JiaqiWang18]

added unit test "consecutive failure event level is correct". flowToNumConsecutiveFailure controls the log level and if it is incremented correctly, we should only see the last retry execution event having ERROR level and all previous events having WARN level .

[AnishMahto]

Just curious, what are we using from the implicits in this test?

[gengliangwang]

It is used in

($"id" % 2).cast("int")

[gengliangwang]

Thanks, merging to master