[PLUGIN-1856] Error management for Wrangler plugin

[Amit-CloudSufi]

https://cdap.atlassian.net/browse/PLUGIN-1856

[
  {
    "stageName": "Wrangler",
    "errorCategory": "Plugin-'Wrangler'",
    "errorReason": "Error in stage 'Wrangler'. Format of output schema specified is invalid. Please check the format. com.google.gson.stream.MalformedJsonException: Unterminated object at line 3 column 4 path $.type",
    "errorMessage": "Error in stage 'Wrangler'. Format of output schema specified is invalid. Please check the format. com.google.gson.stream.MalformedJsonException: Unterminated object at line 3 column 4 path $.type",
    "errorType": "USER",
    "dependency": "false"
  }
]

2025-01-30 12:36:48,237 - ERROR [spark-submitter-phase-1-bfc7d018-ded8-11ef-8496-0000009adf90:o.a.s.i.i.SparkHadoopWriter@98] - Aborting job job_202501301236463168859500037448187_0005.
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 0.0 failed 1 times, most recent failure: Lost task 0.0 in stage 0.0 (TID 0) (192.168.1.100 executor driver): org.apache.spark.SparkException: Task failed while writing rows
	at org.apache.spark.internal.io.SparkHadoopWriter$.executeTask(SparkHadoopWriter.scala:163)
	at org.apache.spark.internal.io.SparkHadoopWriter$.$anonfun$write$1(SparkHadoopWriter.scala:88)
	at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
	at org.apache.spark.scheduler.Task.run(Task.scala:136)
	at org.apache.spark.executor.Executor$TaskRunner.$anonfun$run$3(Executor.scala:548)
	at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1504)
	at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:551)
	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:750)
Caused by: java.util.concurrent.ExecutionException: Error when transforming stage Wrangler: com.google.common.util.concurrent.UncheckedExecutionException: io.cdap.cdap.api.exception.WrappedStageException: Stage 'Wrangler' encountered : io.cdap.cdap.api.exception.ProgramFailureException: Error in stage 'Wrangler'. Format of output schema specified is invalid. Please check the format. com.google.gson.stream.MalformedJsonException: Unterminated object at line 3 column 4 path $.type
	at io.cdap.cdap.etl.spark.function.TransformFunction.call(TransformFunction.java:61)
	at org.apache.spark.api.java.JavaRDDLike.$anonfun$flatMap$1(JavaRDDLike.scala:125)
	at scala.collection.Iterator$$anon$11.nextCur(Iterator.scala:486)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:492)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:491)
	at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:491)
	at org.apache.spark.internal.io.SparkHadoopWriter$.$anonfun$executeTask$1(SparkHadoopWriter.scala:136)
	at org.apache.spark.util.Utils$.tryWithSafeFinallyAndFailureCallbacks(Utils.scala:1538)
	at org.apache.spark.internal.io.SparkHadoopWriter$.executeTask(SparkHadoopWriter.scala:135)
	... 9 more

Loading Directive exception:

Precondition excdeption:

Directive Parsing Exception

RecordConvertorException

[psainics]

Please Fix Unit Tests

[psainics]

E2E Fixed in #727 !

[itsankit-google]

similar comment here

[itsankit-google]

Also this method is private, how was it tested?

[itsankit-google]

LGTM

Please squash commits before merge.

Please remember to cherry-pick the PR in release/4.11.