Skip to content

[SPARK-32175][CORE] Fix the order between initialization for ExecutorPlugin and starting heartbeat thread #29002

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
sarutak wants to merge 8 commits into from
Closed

[SPARK-32175][CORE] Fix the order between initialization for ExecutorPlugin and starting heartbeat thread #29002

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
d724179
Fixed the order of ExecutorPlugin initialization and starting hertbea…
sarutak Jul 5, 2020
1ccb1b7
Tweak the sleep time.
sarutak Jul 8, 2020
9ed108d
Merge branch 'master' of https://github.com/apache/spark into fix-hea…
sarutak Jul 13, 2020
dba4c91
Refactored the code.
sarutak Jul 13, 2020
d768385
Extended timeout for runSparkSubmit.
sarutak Jul 14, 2020
9995ef1
Merge branch 'master' of https://github.com/apache/spark into fix-hea…
sarutak Jul 16, 2020
0b8da96
Make testing time shorter.
sarutak Jul 25, 2020
f106fe9
Merge branch 'master' of https://github.com/apache/spark into fix-hea…
sarutak Jul 29, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
13 changes: 10 additions & 3 deletions core/src/main/scala/org/apache/spark/TestUtils.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,11 +179,18 @@ private[spark] object TestUtils {
destDir: File,
toStringValue: String = "",
baseClass: String = null,
classpathUrls: Seq[URL] = Seq.empty): File = {
classpathUrls: Seq[URL] = Seq.empty,
preClassDefinitionBlock: String = "",
implementsClasses: Seq[String] = Seq.empty,
extraCodeBody: String = ""): File = {
val extendsText = Option(baseClass).map { c => s" extends ${c}" }.getOrElse("")
val implementsText = implementsClasses.map(", " + _).mkString

@Ngone51 Ngone51 Jul 11, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

maybe?

Suggested change
val implementsText = implementsClasses.map(", " + _).mkString
val implementsText =
s"implements ${implementsClasses ++ Seq("java.io.Serializable").mkString(", ")}"

@sarutak sarutak Jul 13, 2020 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks. To be much simpler, I'll make it (implementsClasses :+ "java.io.Serializable").mkString(", ").

val sourceFile = new JavaSourceFromString(className,
"public class " + className + extendsText + " implements java.io.Serializable {" +
" @Override public String toString() { return \"" + toStringValue + "\"; }}")
preClassDefinitionBlock +
"public class " + className + extendsText + " implements java.io.Serializable" +
implementsText + " {" +
" @Override public String toString() { return \"" + toStringValue + "\"; }" +
extraCodeBody + " }")
createCompiledClass(className, destDir, sourceFile, classpathUrls)
}

Expand Down
10 changes: 5 additions & 5 deletions core/src/main/scala/org/apache/spark/executor/Executor.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -154,11 +154,6 @@ private[spark] class Executor(
// for fetching remote cached RDD blocks, so need to make sure it uses the right classloader too.
env.serializerManager.setDefaultClassLoader(replClassLoader)

// Plugins need to load using a class loader that includes the executor's user classpath
private val plugins: Option[PluginContainer] = Utils.withContextClassLoader(replClassLoader) {
PluginContainer(env, resources.asJava)
}

// Max size of direct result. If task result is bigger than this, we use the block manager
// to send the result back.
private val maxDirectResultSize = Math.min(
Expand Down Expand Up @@ -227,6 +222,11 @@ private[spark] class Executor(

metricsPoller.start()
Comment thread
tgravescs marked this conversation as resolved.
Outdated

// Plugins need to load using a class loader that includes the executor's user classpath

@Ngone51 Ngone51 Jul 11, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you add comments to explain why we need to initialize plugin after heartbeater?

private val plugins: Option[PluginContainer] = Utils.withContextClassLoader(replClassLoader) {
PluginContainer(env, resources.asJava)
}

private[executor] def numRunningTasks: Int = runningTasks.size()

/**
Expand Down
76 changes: 74 additions & 2 deletions core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@

package org.apache.spark.executor

import java.io.{Externalizable, ObjectInput, ObjectOutput}
import java.io.{Externalizable, File, ObjectInput, ObjectOutput}
import java.lang.Thread.UncaughtExceptionHandler
import java.nio.ByteBuffer
import java.util.Properties
Expand All @@ -41,6 +41,7 @@ import org.scalatestplus.mockito.MockitoSugar
import org.apache.spark._
import org.apache.spark.TaskState.TaskState
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.deploy.{SimpleApplicationTest, SparkSubmitSuite}
import org.apache.spark.internal.config._
import org.apache.spark.internal.config.UI._
import org.apache.spark.memory.TestMemoryManager
Expand All @@ -52,7 +53,7 @@ import org.apache.spark.scheduler.{DirectTaskResult, FakeTask, ResultTask, Task,
import org.apache.spark.serializer.{JavaSerializer, SerializerInstance, SerializerManager}
import org.apache.spark.shuffle.FetchFailedException
import org.apache.spark.storage.{BlockManager, BlockManagerId}
import org.apache.spark.util.{LongAccumulator, UninterruptibleThread}
import org.apache.spark.util.{LongAccumulator, UninterruptibleThread, Utils}

class ExecutorSuite extends SparkFunSuite
with LocalSparkContext with MockitoSugar with Eventually with PrivateMethodTester {
Expand Down Expand Up @@ -402,6 +403,77 @@ class ExecutorSuite extends SparkFunSuite
assert(taskMetrics.getMetricValue("JVMHeapMemory") > 0)
}

test("SPARK-32175: Plugin initialization should start after heartbeater started") {
val tempDir = Utils.createTempDir()

@Ngone51 Ngone51 Jul 11, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

clean it up at the end of the test? (though I know it will be cleaned by shutdown hook anyway.)

@sarutak sarutak Jul 13, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll try to use withTempDir.


val importStatements =
"""
|import java.util.Map;
|import org.apache.spark.api.plugin.*;

@Ngone51 Ngone51 Jul 11, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What about using the qualified class name to avoid adding the new parameter preClassDefinitionBlock?

@sarutak sarutak Jul 13, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, it's better.

""".stripMargin
val sparkPluginCodeBody =
"""
|@Override
|public ExecutorPlugin executorPlugin() {
| return new TestExecutorPlugin();
|}
|
|@Override
|public DriverPlugin driverPlugin() { return null; }
""".stripMargin
val executorPluginBody =
"""
|@Override
|public void init(PluginContext ctx, Map<String, String> extraConf) {
| try {
| Thread.sleep(30 * 1000);
| } catch (InterruptedException e) {
| throw new RuntimeException(e);
| }
|}
""".stripMargin

val compiledExecutorPlugin = TestUtils.createCompiledClass(
"TestExecutorPlugin",
tempDir,
"",
null,
Seq.empty,
importStatements,
Seq("ExecutorPlugin"),
executorPluginBody)

val thisClassPath =
sys.props("java.class.path").split(File.pathSeparator).map(p => new File(p).toURI.toURL)
val compiledSparkPlugin = TestUtils.createCompiledClass(
"TestSparkPlugin",
tempDir,
"",
null,
Seq(tempDir.toURI.toURL) ++ thisClassPath,
importStatements,
Seq("SparkPlugin"),
sparkPluginCodeBody)

val jarUrl = TestUtils.createJar(
Seq(compiledSparkPlugin, compiledExecutorPlugin),
new File(tempDir, "testPlugin.jar"))

val unusedJar = TestUtils.createJarWithClasses(Seq.empty)
val args = Seq(
"--class", SimpleApplicationTest.getClass.getName.stripSuffix("$"),
"--name", "testApp",
"--master", "local-cluster[1,1,1024]",
"--conf", "spark.plugins=TestSparkPlugin",
"--conf", "spark.storage.blockManagerSlaveTimeoutMs=" + 10 * 1000,
"--conf", "spark.network.timeoutInterval=" + 10 * 1000,
"--conf", "spark.executor.extraClassPath=" + jarUrl.toString,
"--conf", "spark.driver.extraClassPath=" + jarUrl.toString,
"--conf", "spark.ui.enabled=false",
unusedJar.toString)
SparkSubmitSuite.runSparkSubmit(args)
}

private def createMockEnv(conf: SparkConf, serializer: JavaSerializer): SparkEnv = {
val mockEnv = mock[SparkEnv]
val mockRpcEnv = mock[RpcEnv]
Expand Down