Merge branch 'master' into spark-34152

Author: imback82

core/src/main/resources/org/apache/spark/ui/static/executorspage-template.html

@@ -104,7 +104,7 @@ <h4 class="title-table">Executors</h4>
               Peak Pool Memory Direct / Mapped</span></th>
           <th>Disk Used</th>
           <th>Cores</th>
-          <th>Resources</th>
+          <th><span data-toggle="tooltip" data-placement="top" title="Resources currently used by each executor, such as GPU, FPGA, etc.">Resources</span></th>
           <th>Resource Profile Id</th>
           <th><span data-toggle="tooltip" data-placement="top" title="Number of tasks currently executing. Darker shading highlights executors with more active tasks.">Active Tasks</span></th>
           <th><span data-toggle="tooltip" data-placement="top" title="Number of tasks that have failed on this executor. Darker shading highlights executors with a high proportion of failed tasks.">Failed Tasks</span></th>

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

@@ -128,14 +128,16 @@ private[spark] class HeartbeatReceiver(sc: SparkContext, clock: Clock)
 
     // Messages received from executors
     case heartbeat @ Heartbeat(executorId, accumUpdates, blockManagerId, executorUpdates) =>
+      var reregisterBlockManager = !sc.isStopped
       if (scheduler != null) {
         if (executorLastSeen.contains(executorId)) {
           executorLastSeen(executorId) = clock.getTimeMillis()
           eventLoopThread.submit(new Runnable {
             override def run(): Unit = Utils.tryLogNonFatalError {
               val unknownExecutor = !scheduler.executorHeartbeatReceived(
                 executorId, accumUpdates, blockManagerId, executorUpdates)
-              val response = HeartbeatResponse(reregisterBlockManager = unknownExecutor)
+              reregisterBlockManager &= unknownExecutor
+              val response = HeartbeatResponse(reregisterBlockManager)
               context.reply(response)
             }
           })
@@ -145,14 +147,14 @@ private[spark] class HeartbeatReceiver(sc: SparkContext, clock: Clock)
           // not log warning here. Otherwise there may be a lot of noise especially if
           // we explicitly remove executors (SPARK-4134).
           logDebug(s"Received heartbeat from unknown executor $executorId")
-          context.reply(HeartbeatResponse(reregisterBlockManager = true))
+          context.reply(HeartbeatResponse(reregisterBlockManager))
         }
       } else {
         // Because Executor will sleep several seconds before sending the first "Heartbeat", this
         // case rarely happens. However, if it really happens, log it and ask the executor to
         // register itself again.
         logWarning(s"Dropping $heartbeat because TaskScheduler is not ready yet")
-        context.reply(HeartbeatResponse(reregisterBlockManager = true))
+        context.reply(HeartbeatResponse(reregisterBlockManager))
       }
   }
 

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

@@ -219,6 +219,24 @@ class HeartbeatReceiverSuite
     fakeSchedulerBackend.stop()
   }
 
+  test("SPARK-34273: Do not reregister BlockManager when SparkContext is stopped") {
+    val blockManagerId = BlockManagerId(executorId1, "localhost", 12345)
+
+    heartbeatReceiverRef.askSync[Boolean](TaskSchedulerIsSet)
+    val response = heartbeatReceiverRef.askSync[HeartbeatResponse](
+      Heartbeat(executorId1, Array.empty, blockManagerId, mutable.Map.empty))
+    assert(response.reregisterBlockManager)
+
+    try {
+      sc.stopped.set(true)
+      val response = heartbeatReceiverRef.askSync[HeartbeatResponse](
+        Heartbeat(executorId1, Array.empty, blockManagerId, mutable.Map.empty))
+      assert(!response.reregisterBlockManager)
+    } finally {
+      sc.stopped.set(false)
+    }
+  }
+
   /** Manually send a heartbeat and return the response. */
   private def triggerHeartbeat(
       executorId: String,

core/src/test/scala/org/apache/spark/deploy/SparkSubmitSuite.scala

@@ -1106,8 +1106,8 @@ class SparkSubmitSuite
 
     // The path and filename are preserved.
     assert(outputUri.getPath.endsWith(new Path(sourceUri).getName))
-    assert(FileUtils.readFileToString(new File(outputUri.getPath)) ===
-      FileUtils.readFileToString(new File(sourceUri.getPath)))
+    assert(FileUtils.readFileToString(new File(outputUri.getPath), StandardCharsets.UTF_8) ===
+      FileUtils.readFileToString(new File(sourceUri.getPath), StandardCharsets.UTF_8))
   }
 
   private def deleteTempOutputFile(outputPath: String): Unit = {
@@ -1149,7 +1149,7 @@ class SparkSubmitSuite
     val jarFile = File.createTempFile("test", ".jar")
     jarFile.deleteOnExit()
     val content = "hello, world"
-    FileUtils.write(jarFile, content)
+    FileUtils.write(jarFile, content, StandardCharsets.UTF_8)
     val hadoopConf = new Configuration()
     val tmpDir = Files.createTempDirectory("tmp").toFile
     updateConfWithFakeS3Fs(hadoopConf)
@@ -1164,7 +1164,7 @@ class SparkSubmitSuite
     val jarFile = File.createTempFile("test", ".jar")
     jarFile.deleteOnExit()
     val content = "hello, world"
-    FileUtils.write(jarFile, content)
+    FileUtils.write(jarFile, content, StandardCharsets.UTF_8)
     val hadoopConf = new Configuration()
     val tmpDir = Files.createTempDirectory("tmp").toFile
     updateConfWithFakeS3Fs(hadoopConf)

core/src/test/scala/org/apache/spark/deploy/history/HistoryServerSuite.scala

@@ -199,7 +199,8 @@ class HistoryServerSuite extends SparkFunSuite with BeforeAndAfter with Matchers
       errOpt should be (None)
 
       val exp = IOUtils.toString(new FileInputStream(
-        new File(expRoot, HistoryServerSuite.sanitizePath(name) + "_expectation.json")))
+        new File(expRoot, HistoryServerSuite.sanitizePath(name) + "_expectation.json")),
+        StandardCharsets.UTF_8)
       // compare the ASTs so formatting differences don't cause failures
       import org.json4s._
       import org.json4s.jackson.JsonMethods._
@@ -713,7 +714,7 @@ object HistoryServerSuite {
 
   def getContentAndCode(url: URL): (Int, Option[String], Option[String]) = {
     val (code, in, errString) = connectAndGetInputStream(url)
-    val inString = in.map(IOUtils.toString)
+    val inString = in.map(IOUtils.toString(_, StandardCharsets.UTF_8))
     (code, inString, errString)
   }
 
@@ -729,7 +730,7 @@ object HistoryServerSuite {
     }
     val errString = try {
       val err = Option(connection.getErrorStream())
-      err.map(IOUtils.toString)
+      err.map(IOUtils.toString(_, StandardCharsets.UTF_8))
     } catch {
       case io: IOException => None
     }

dev/deps/spark-deps-hadoop-2.7-hive-2.3

@@ -202,12 +202,12 @@ orc-shims/1.6.7//orc-shims-1.6.7.jar
 oro/2.0.8//oro-2.0.8.jar
 osgi-resource-locator/1.0.3//osgi-resource-locator-1.0.3.jar
 paranamer/2.8//paranamer-2.8.jar
-parquet-column/1.10.1//parquet-column-1.10.1.jar
-parquet-common/1.10.1//parquet-common-1.10.1.jar
-parquet-encoding/1.10.1//parquet-encoding-1.10.1.jar
-parquet-format/2.4.0//parquet-format-2.4.0.jar
-parquet-hadoop/1.10.1//parquet-hadoop-1.10.1.jar
-parquet-jackson/1.10.1//parquet-jackson-1.10.1.jar
+parquet-column/1.11.1//parquet-column-1.11.1.jar
+parquet-common/1.11.1//parquet-common-1.11.1.jar
+parquet-encoding/1.11.1//parquet-encoding-1.11.1.jar
+parquet-format-structures/1.11.1//parquet-format-structures-1.11.1.jar
+parquet-hadoop/1.11.1//parquet-hadoop-1.11.1.jar
+parquet-jackson/1.11.1//parquet-jackson-1.11.1.jar
 protobuf-java/2.5.0//protobuf-java-2.5.0.jar
 py4j/0.10.9.1//py4j-0.10.9.1.jar
 pyrolite/4.30//pyrolite-4.30.jar

dev/deps/spark-deps-hadoop-3.2-hive-2.3

@@ -172,12 +172,12 @@ orc-shims/1.6.7//orc-shims-1.6.7.jar
 oro/2.0.8//oro-2.0.8.jar
 osgi-resource-locator/1.0.3//osgi-resource-locator-1.0.3.jar
 paranamer/2.8//paranamer-2.8.jar
-parquet-column/1.10.1//parquet-column-1.10.1.jar
-parquet-common/1.10.1//parquet-common-1.10.1.jar
-parquet-encoding/1.10.1//parquet-encoding-1.10.1.jar
-parquet-format/2.4.0//parquet-format-2.4.0.jar
-parquet-hadoop/1.10.1//parquet-hadoop-1.10.1.jar
-parquet-jackson/1.10.1//parquet-jackson-1.10.1.jar
+parquet-column/1.11.1//parquet-column-1.11.1.jar
+parquet-common/1.11.1//parquet-common-1.11.1.jar
+parquet-encoding/1.11.1//parquet-encoding-1.11.1.jar
+parquet-format-structures/1.11.1//parquet-format-structures-1.11.1.jar
+parquet-hadoop/1.11.1//parquet-hadoop-1.11.1.jar
+parquet-jackson/1.11.1//parquet-jackson-1.11.1.jar
 protobuf-java/2.5.0//protobuf-java-2.5.0.jar
 py4j/0.10.9.1//py4j-0.10.9.1.jar
 pyrolite/4.30//pyrolite-4.30.jar

docs/running-on-kubernetes.md

@@ -193,6 +193,9 @@ for any reason, these pods will remain in the cluster. The executor processes sh
 driver, so the executor pods should not consume compute resources (cpu and memory) in the cluster after your application
 exits.
 
+You may use `spark.kubernetes.executor.podNamePrefix` to fully control the executor pod names.
+When this property is set, it's highly recommended to make it unique across all jobs in the same namespace.
+
 ### Authentication Parameters
 
 Use the exact prefix `spark.kubernetes.authenticate` for Kubernetes authentication parameters in client mode.
@@ -873,6 +876,14 @@ See the [configuration page](configuration.html) for information on Spark config
   </td>
   <td>2.3.0</td>
 </tr>
+<tr>
+  <td><code>spark.kubernetes.executor.podNamePrefix</code></td>
+  <td>(none)</td>
+  <td>
+    Prefix to use in front of the executor pod names.
+  </td>
+  <td>2.3.0</td>
+</tr>
 <tr>
   <td><code>spark.kubernetes.executor.lostCheck.maxAttempts</code></td>
   <td><code>10</code></td>

pom.xml

@@ -136,7 +136,7 @@
     <kafka.version>2.6.0</kafka.version>
     <!-- After 10.15.1.3, the minimum required version is JDK9 -->
     <derby.version>10.14.2.0</derby.version>
-    <parquet.version>1.10.1</parquet.version>
+    <parquet.version>1.11.1</parquet.version>
     <orc.version>1.6.7</orc.version>
     <jetty.version>9.4.34.v20201102</jetty.version>
     <jakartaservlet.version>4.0.3</jakartaservlet.version>
@@ -2290,6 +2290,10 @@
             <groupId>commons-pool</groupId>
             <artifactId>commons-pool</artifactId>
           </exclusion>
+          <exclusion>
+            <groupId>javax.annotation</groupId>
+            <artifactId>javax.annotation-api</artifactId>
+          </exclusion>
         </exclusions>
       </dependency>
       <dependency>

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/Config.scala

@@ -224,7 +224,6 @@ private[spark] object Config extends Logging {
     ConfigBuilder("spark.kubernetes.executor.podNamePrefix")
       .doc("Prefix to use in front of the executor pod names.")
       .version("2.3.0")
-      .internal()
       .stringConf
       .createOptional