[KYUUBI #2922] Clean up SparkConsoleProgressBar when SQL execution fails

### _Why are the changes needed?_
Now `SparkConsoleProgressBar` is initialized after constructing `SQLOperationListener`.
Sometimes the user's SQL syntax may fail.
At this time, because there is no `SparkListenerSQLExecutionEnd` event, `SparkConsoleProgressBar` is not cleaned up, resulting in a memory leak.

In this PR, make the following changes

- Added cleanup method

- Some objects use lazy to avoid constructing when SQL syntax is wrong

close #2922

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [x] [Run test](https://kyuubi.apache.org/docs/latest/develop_tools/testing.html#running-tests) locally before make a pull request

Closes #2923 from cxzl25/KYUUBI-2922.

Closes #2922

42dbdf2 [sychen] fix leak

Authored-by: sychen <sychen@ctrip.com>
Signed-off-by: Shaoyun Chen <csy@apache.org>

Author: cxzl25

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/ExecuteStatement.scala

@@ -144,7 +144,7 @@ class ExecuteStatement(
   }
 
   override def cleanup(targetState: OperationState): Unit = {
-    operationListener.foreach(spark.sparkContext.removeSparkListener(_))
+    operationListener.foreach(_.cleanup())
     super.cleanup(targetState)
   }
 

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/spark/kyuubi/SQLOperationListener.scala

@@ -44,13 +44,13 @@ class SQLOperationListener(
     spark: SparkSession) extends StatsReportListener with Logging {
 
   private val operationId: String = operation.getHandle.identifier.toString
-  private val activeJobs = new java.util.HashSet[Int]()
-  private val activeStages = new java.util.HashSet[Int]()
+  private lazy val activeJobs = new java.util.HashSet[Int]()
+  private lazy val activeStages = new java.util.HashSet[Int]()
   private var executionId: Option[Long] = None
-  private val liveStages = new ConcurrentHashMap[StageAttempt, StageInfo]()
+  private lazy val liveStages = new ConcurrentHashMap[StageAttempt, StageInfo]()
 
   private val conf: KyuubiConf = operation.getSession.sessionManager.getConf
-  private val consoleProgressBar =
+  private lazy val consoleProgressBar =
     if (conf.get(ENGINE_SPARK_SHOW_PROGRESS)) {
       Some(new SparkConsoleProgressBar(
         operation,
@@ -87,6 +87,7 @@ class SQLOperationListener(
       if (executionId.isEmpty) {
         executionId = Option(jobStart.properties.getProperty(SPARK_SQL_EXECUTION_ID_KEY))
           .map(_.toLong)
+        consoleProgressBar
         operation match {
           case executeStatement: ExecuteStatement =>
             executeStatement.setCompiledStateIfNeeded()
@@ -163,9 +164,15 @@ class SQLOperationListener(
     event match {
       case sqlExecutionEnd: SparkListenerSQLExecutionEnd
           if executionId.contains(sqlExecutionEnd.executionId) =>
-        spark.sparkContext.removeSparkListener(this)
-        consoleProgressBar.foreach(_.finish())
+        cleanup()
       case _ =>
     }
   }
+
+  def cleanup(): Unit = {
+    spark.sparkContext.removeSparkListener(this)
+    if (executionId.isDefined) {
+      consoleProgressBar.foreach(_.finish())
+    }
+  }
 }