address comments

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

@@ -2698,10 +2698,8 @@ class SparkContext(config: SparkConf) extends Logging {
    * @param shouldCancelJob Callback function to be called with the job ID of each job that matches
    *    the given tag. If the function returns true, the job will be cancelled.
    * @return A future that will be completed with the set of job IDs that were cancelled.
-   *
-   * @since 4.0.0
    */
-  def cancelJobsWithTag(
+  private[spark] def cancelJobsWithTag(
       tag: String,
       reason: String,
       shouldCancelJob: ActiveJob => Boolean): Future[Set[Int]] = {
@@ -2755,7 +2753,7 @@ class SparkContext(config: SparkConf) extends Logging {
    *    the given tag. If the function returns true, the job will be cancelled.
    * @return A future that will be completed with the set of job IDs that were cancelled.
    */
-  def cancelAllJobs(shouldCancelJob: ActiveJob => Boolean): Future[Set[Int]] = {
+  private[spark] def cancelAllJobs(shouldCancelJob: ActiveJob => Boolean): Future[Set[Int]] = {
     assertNotStopped()
 
     val cancelledJobs = Promise[Set[Int]]()
@@ -2779,7 +2777,7 @@ class SparkContext(config: SparkConf) extends Logging {
    * @return A future that will be completed with the set of job IDs that were cancelled.
    * @note Throws `InterruptedException` if the cancel message cannot be sent
    */
-  def cancelJob(
+  private[spark] def cancelJob(
       jobId: Int,
       reason: String,
       shouldCancelJob: ActiveJob => Boolean): Future[Set[Int]] = {

sql/core/src/main/scala/org/apache/spark/sql/SparkSession.scala

@@ -93,7 +93,8 @@ class SparkSession private(
     @transient private val existingSharedState: Option[SharedState],
     @transient private val parentSessionState: Option[SessionState],
     @transient private[sql] val extensions: SparkSessionExtensions,
-    @transient private[sql] val initialSessionOptions: Map[String, String])
+    @transient private[sql] val initialSessionOptions: Map[String, String],
+    @transient private val parentUserDefinedToRealTagsMap: Map[String, String])
   extends Serializable with Closeable with Logging { self =>
 
   // The call site where this SparkSession was constructed.
@@ -108,8 +109,13 @@ class SparkSession private(
   private[sql] def this(
       sc: SparkContext,
       initialSessionOptions: java.util.HashMap[String, String]) = {
-    this(sc, None, None, SparkSession.applyExtensions(sc, new SparkSessionExtensions),
-      initialSessionOptions.asScala.toMap)
+    this(
+      sc,
+      existingSharedState = None,
+      parentSessionState = None,
+      SparkSession.applyExtensions(sc, new SparkSessionExtensions),
+      initialSessionOptions.asScala.toMap,
+      parentUserDefinedToRealTagsMap = Map.empty)
   }
 
   private[sql] def this(sc: SparkContext) = this(sc, new java.util.HashMap[String, String]())
@@ -128,7 +134,9 @@ class SparkSession private(
    * A map to hold the mapping from user-defined tags to the real tags attached to Jobs.
    * Real tag have the current session ID attached: `"tag1" -> s"spark-$sessionUUID-tag1"`.
    */
-  private val userDefinedToRealTagsMap: ConcurrentHashMap[String, String] = new ConcurrentHashMap()
+  @transient
+  private lazy val userDefinedToRealTagsMap: ConcurrentHashMap[String, String] =
+    new ConcurrentHashMap(parentUserDefinedToRealTagsMap.asJava)
 
   /**
    * The version of Spark on which this application is running.
@@ -285,7 +293,8 @@ class SparkSession private(
       Some(sharedState),
       parentSessionState = None,
       extensions,
-      initialSessionOptions)
+      initialSessionOptions,
+      parentUserDefinedToRealTagsMap = Map.empty)
   }
 
   /**
@@ -306,8 +315,10 @@ class SparkSession private(
       Some(sharedState),
       Some(sessionState),
       extensions,
-      Map.empty)
+      Map.empty,
+      userDefinedToRealTagsMap.asScala.toMap)
     result.sessionState // force copy of SessionState
+    result.userDefinedToRealTagsMap // force copy of userDefinedToRealTagsMap
     result
   }
 
@@ -1261,7 +1272,12 @@ object SparkSession extends Logging {
         loadExtensions(extensions)
         applyExtensions(sparkContext, extensions)
 
-        session = new SparkSession(sparkContext, None, None, extensions, options.toMap)
+        session = new SparkSession(sparkContext,
+          existingSharedState = None,
+          parentSessionState = None,
+          extensions,
+          initialSessionOptions = options.toMap,
+          parentUserDefinedToRealTagsMap = Map.empty)
         setDefaultSession(session)
         setActiveSession(session)
         registerContextListener(sparkContext)

sql/core/src/test/scala/org/apache/spark/sql/SparkSessionJobCancellationSuite.scala → sql/core/src/test/scala/org/apache/spark/sql/SparkSessionJobTaggingAndCancellationSuite.scala

@@ -39,10 +39,10 @@ import org.apache.spark.tags.ExtendedSQLTest
 import org.apache.spark.util.ThreadUtils
 
 /**
- * Test cases for the cancellation APIs provided by [[SparkSession]].
+ * Test cases for the tagging and cancellation APIs provided by [[SparkSession]].
  */
 @ExtendedSQLTest
-class SparkSessionJobCancellationSuite
+class SparkSessionJobTaggingAndCancellationSuite
   extends SparkFunSuite
   with Eventually
   with LocalSparkContext {
@@ -60,6 +60,33 @@ class SparkSessionJobCancellationSuite
     }
   }
 
+  test("Tags are not inherited by new sessions") {
+    val session = SparkSession.builder().master("local").getOrCreate()
+
+    assert(session.getTags() == Set())
+    session.addTag("one")
+    assert(session.getTags() == Set("one"))
+
+    val newSession = session.newSession()
+    assert(newSession.getTags() == Set())
+  }
+
+  test("Tags are inherited by cloned sessions") {
+    val session = SparkSession.builder().master("local").getOrCreate()
+
+    assert(session.getTags() == Set())
+    session.addTag("one")
+    assert(session.getTags() == Set("one"))
+
+    val clonedSession = session.cloneSession()
+    assert(clonedSession.getTags() == Set("one"))
+    clonedSession.addTag("two")
+    assert(clonedSession.getTags() == Set("one", "two"))
+
+    // Tags are not propagated back to the original session
+    assert(session.getTags() == Set("one"))
+  }
+
   test("Cancellation APIs in SparkSession are isolated") {
     sc = new SparkContext("local[2]", "test")
     val globalSession = SparkSession.builder().sparkContext(sc).getOrCreate()