[SPARK-51095][CORE][SQL] Include caller context for hdfs audit logs for calls from driver

### What changes were proposed in this pull request?

Add the caller context for calls from DRIVER to HDFS.

### Why are the changes needed?
HDFS audit logs include the ability to add a "caller context".  Spark already leverages this to set the yarn application id, job id, task id, etc. but only on executors.  The caller context is left empty on the spark driver.  With introductions of Iceberg we have seen multiple scenarios in which files in HDFS are accessed from the driver. But since the caller context is left empty our ability to forensically analyse any issues has diminished. This PR includes sets caller context from the driver as well.

### Does this PR introduce _any_ user-facing change?
Yes, hdfs audit logs will now have caller context for calls from driver.

### How was this patch tested?
This patch was tested manually. After this change the hdfs audit logs now contain caller context from the driver.

```
2025-02-14 02:26:23,249 INFO FSNamesystem.audit: allowed=true   ugi=root (auth:SIMPLE)  ip=/192.168.97.4        cmd=getfileinfo src=/warehouse/sample   dst=null        perm=null       proto=rpc       callerContext=SPARK_DRIVER_application_1739496632907_0005
2025-02-14 02:26:23,265 INFO FSNamesystem.audit: allowed=true   ugi=root (auth:SIMPLE)  ip=/192.168.97.4        cmd=listStatus  src=/warehouse/sample   dst=null        perm=null       proto=rpc       callerContext=SPARK_DRIVER_application_1739496632907_0005
2025-02-14 02:26:25,519 INFO FSNamesystem.audit: allowed=true   ugi=root (auth:SIMPLE)  ip=/192.168.97.5        cmd=open        src=/warehouse/sample/part-00000-dd473344-76b1-4179-91ae-d15a8da4a888-c000      dst=null        perm=null       proto=rpc       callerContext=SPARK_TASK_application_1739496632907_0005_JId_0_SId_0_0_TId_0_0
2025-02-14 02:26:26,345 INFO FSNamesystem.audit: allowed=true   ugi=root (auth:SIMPLE)  ip=/192.168.97.5        cmd=open        src=/warehouse/sample/part-00000-dd473344-76b1-4179-91ae-d15a8da4a888-c000      dst=null        perm=null       proto=rpc       callerContext=SPARK_TASK_application_1739496632907_0005_JId_1_SId_1_0_TId_1_0
```

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

No

Closes apache#49814 from sririshindra/master-SPARK-51095.

Lead-authored-by: Rishi <[email protected]>
Co-authored-by: Rishi <[email protected]>
Signed-off-by: attilapiros <[email protected]>

Author: sririshindra

core/src/main/scala/org/apache/spark/SparkContext.scala

@@ -722,6 +722,9 @@ class SparkContext(config: SparkConf) extends Logging {
     }
     appStatusSource.foreach(_env.metricsSystem.registerSource(_))
     _plugins.foreach(_.registerMetrics(applicationId))
+
+    new CallerContext("DRIVER", config.get(APP_CALLER_CONTEXT),
+      Some(applicationId), applicationAttemptId).setCurrentContext()
   } catch {
     case NonFatal(e) =>
       logError("Error initializing SparkContext.", e)

core/src/test/scala/org/apache/spark/SparkContextSuite.scala

@@ -29,6 +29,7 @@ import com.google.common.io.Files
 import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.fs.Path
 import org.apache.hadoop.io.{BytesWritable, LongWritable, Text}
+import org.apache.hadoop.ipc.{CallerContext => HadoopCallerContext}
 import org.apache.hadoop.mapred.TextInputFormat
 import org.apache.hadoop.mapreduce.lib.input.{TextInputFormat => NewTextInputFormat}
 import org.apache.logging.log4j.{Level, LogManager}
@@ -1460,6 +1461,15 @@ class SparkContextSuite extends SparkFunSuite with LocalSparkContext with Eventu
     }
     sc.stop()
   }
+
+  test("SPARK-51095: Test caller context initialization") {
+    val conf = new SparkConf().setAppName("test").setMaster("local")
+    sc = new SparkContext(conf)
+    val hadoopCallerContext = HadoopCallerContext.getCurrent()
+    assert(hadoopCallerContext.getContext().startsWith("SPARK_DRIVER"))
+    sc.stop()
+  }
+
 }
 
 object SparkContextSuite {