You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@livy.apache.org by "nileshrathi345 (via GitHub)" <gi...@apache.org> on 2026/07/24 06:44:13 UTC
[PR] LIVY-866: Optimize Yarn GetApplications Query to prevent additional load on Yarn and Livy [livy]
nileshrathi345 opened a new pull request, #533:
URL: https://github.com/apache/livy/pull/533
## What changes were proposed in this pull request?
**Problem**
Livy was calling YarnClient.getApplications(SPARK), which asks the ResourceManager for all Spark applications and then filters client-side by application tag. On clusters with many Spark apps, this produces large RM responses, adds unnecessary load on YARN, and can contribute to RM memory pressure (especially during session startup polling and leaked-app GC).
**Solution**
Use a tag-filtered GetApplicationsRequest so the RM returns only matching applications:
- **getAppIdFromTag()** — when resolving a session’s YARN app ID by tag, query with the lowercase session tag instead of scanning all Spark apps.
- **leakedAppsGCThread** — when cleaning up leaked apps, batch all pending leaked tags into one filtered request per GC cycle instead of fetching the full Spark app list.
Jira: https://issues.apache.org/jira/browse/LIVY-866
## How was this patch tested?
Updated and ran SparkYarnAppSpec, including:
- "should get App Id" — verifies tag-based app ID lookup uses the filtered getApplications request.
- "should delete leak app when timeout" — verifies the leaked-app GC path uses the filtered request.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@livy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] LIVY-866: Optimize Yarn GetApplications Query to prevent additional load on Yarn and Livy [livy]
Posted by "gyogal (via GitHub)" <gi...@apache.org>.
gyogal merged PR #533:
URL: https://github.com/apache/livy/pull/533
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@livy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] LIVY-866: Optimize Yarn GetApplications Query to prevent additional load on Yarn and Livy [livy]
Posted by "nileshrathi345 (via GitHub)" <gi...@apache.org>.
nileshrathi345 commented on PR #533:
URL: https://github.com/apache/livy/pull/533#issuecomment-5090365528
@gyogal, @ArnavBalyan and @roczei could you please take a look when possible and review
thanks..!!!!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@livy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] LIVY-866: Optimize Yarn GetApplications Query to prevent additional load on Yarn and Livy [livy]
Posted by "gyogal (via GitHub)" <gi...@apache.org>.
gyogal commented on code in PR #533:
URL: https://github.com/apache/livy/pull/533#discussion_r3693565773
##########
server/src/main/scala/org/apache/livy/utils/SparkYarnApp.scala:
##########
@@ -68,27 +71,40 @@ object SparkYarnApp extends Logging {
private var sessionLeakageCheckInterval: Long = _
+ /**
+ * Build a GetApplicationsRequest filtered by Spark application type and tags.
+ * Tags are normalized to lowercase to match YARN's tag storage behavior.
+ */
+ private def createGetApplicationsRequest(appTags: util.Set[String]): GetApplicationsRequest = {
+ val normalizedTags = new util.HashSet[String]()
+ appTags.asScala.foreach(tag => normalizedTags.add(tag.toLowerCase))
+ val request = GetApplicationsRequest.newInstance(appType)
+ request.setApplicationTags(normalizedTags)
+ request
+ }
+
private val leakedAppsGCThread = new Thread() {
override def run(): Unit = {
- val client = {
- mockYarnClient match {
+ while (true) {
+ val client = mockYarnClient match {
Review Comment:
Could this be evaluated once, before the loop?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@livy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] LIVY-866: Optimize Yarn GetApplications Query to prevent additional load on Yarn and Livy [livy]
Posted by "nileshrathi345 (via GitHub)" <gi...@apache.org>.
nileshrathi345 commented on code in PR #533:
URL: https://github.com/apache/livy/pull/533#discussion_r3665038051
##########
server/src/main/scala/org/apache/livy/utils/SparkYarnApp.scala:
##########
@@ -35,13 +38,22 @@ import org.apache.livy.{LivyConf, Logging, Utils}
object SparkYarnApp extends Logging {
+ @volatile private var leakedAppsGCThreadStarted = false
+
def init(livyConf: LivyConf, client: Option[YarnClient] = None): Unit = {
mockYarnClient = client
sessionLeakageCheckInterval = livyConf.getTimeAsMs(LivyConf.YARN_APP_LEAKAGE_CHECK_INTERVAL)
sessionLeakageCheckTimeout = livyConf.getTimeAsMs(LivyConf.YARN_APP_LEAKAGE_CHECK_TIMEOUT)
- leakedAppsGCThread.setDaemon(true)
- leakedAppsGCThread.setName("LeakedAppsGCThread")
- leakedAppsGCThread.start()
+ if (!leakedAppsGCThreadStarted) {
+ synchronized {
Review Comment:
The synchronized block was added to just to make sure SparkYarnApp.init() is safe to call more than once for the unit tests.
I have removed it and restore the previous behavior
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@livy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] LIVY-866: Optimize Yarn GetApplications Query to prevent additional load on Yarn and Livy [livy]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #533:
URL: https://github.com/apache/livy/pull/533#issuecomment-5106753524
## [Codecov](https://app.codecov.io/gh/apache/livy/pull/533?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 68.67%. Comparing base ([`0efa2d7`](https://app.codecov.io/gh/apache/livy/commit/0efa2d7e508c6d955cebe48a5159ffc3202e57a1?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`d6dc967`](https://app.codecov.io/gh/apache/livy/commit/d6dc967ffda8d496c0feac82f3aa56e5923e6c3d?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 31 commits behind head on master.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## master #533 +/- ##
============================================
+ Coverage 68.38% 68.67% +0.29%
- Complexity 1199 1214 +15
============================================
Files 106 106
Lines 6711 6793 +82
Branches 831 839 +8
============================================
+ Hits 4589 4665 +76
- Misses 1657 1661 +4
- Partials 465 467 +2
```
</details>
[:umbrella: View full report in Codecov by Harness](https://app.codecov.io/gh/apache/livy/pull/533?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@livy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] LIVY-866: Optimize Yarn GetApplications Query to prevent additional load on Yarn and Livy [livy]
Posted by "nileshrathi345 (via GitHub)" <gi...@apache.org>.
nileshrathi345 commented on code in PR #533:
URL: https://github.com/apache/livy/pull/533#discussion_r3662731238
##########
server/src/test/scala/org/apache/livy/utils/SparkYarnAppSpec.scala:
##########
@@ -49,6 +50,13 @@ class SparkYarnAppSpec extends FunSpec with LivyBaseUnitTestSuite {
Thread.`yield`()
}
+ private def mockGetApplicationsByTags(
+ client: YarnClient,
+ reports: List[ApplicationReport]): Unit = {
+ when(client.getApplications(any(classOf[GetApplicationsRequest])))
Review Comment:
Stronger tests with ArgumentCaptor — added verifyFilteredGetApplicationsRequest() to assert:
- Request includes expected tags (lowercase)
- Request filters by SPARK application type
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@livy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] LIVY-866: Optimize Yarn GetApplications Query to prevent additional load on Yarn and Livy [livy]
Posted by "nileshrathi345 (via GitHub)" <gi...@apache.org>.
nileshrathi345 commented on code in PR #533:
URL: https://github.com/apache/livy/pull/533#discussion_r3713402895
##########
server/src/main/scala/org/apache/livy/utils/SparkYarnApp.scala:
##########
@@ -68,27 +71,40 @@ object SparkYarnApp extends Logging {
private var sessionLeakageCheckInterval: Long = _
+ /**
+ * Build a GetApplicationsRequest filtered by Spark application type and tags.
+ * Tags are normalized to lowercase to match YARN's tag storage behavior.
+ */
+ private def createGetApplicationsRequest(appTags: util.Set[String]): GetApplicationsRequest = {
+ val normalizedTags = new util.HashSet[String]()
+ appTags.asScala.foreach(tag => normalizedTags.add(tag.toLowerCase))
+ val request = GetApplicationsRequest.newInstance(appType)
+ request.setApplicationTags(normalizedTags)
+ request
+ }
+
private val leakedAppsGCThread = new Thread() {
override def run(): Unit = {
- val client = {
- mockYarnClient match {
+ while (true) {
+ val client = mockYarnClient match {
Review Comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@livy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] LIVY-866: Optimize Yarn GetApplications Query to prevent additional load on Yarn and Livy [livy]
Posted by "roczei (via GitHub)" <gi...@apache.org>.
roczei commented on code in PR #533:
URL: https://github.com/apache/livy/pull/533#discussion_r3656956831
##########
server/src/main/scala/org/apache/livy/utils/SparkYarnApp.scala:
##########
@@ -68,6 +71,15 @@ object SparkYarnApp extends Logging {
private var sessionLeakageCheckInterval: Long = _
+ /**
+ * Build a GetApplicationsRequest filtered by Spark application type and tags.
+ */
+ private def createGetApplicationsRequest(appTags: util.Set[String]): GetApplicationsRequest = {
Review Comment:
Pulling createGetApplicationsRequest() into a private helper is a nice idea to avoid duplication.
##########
server/src/main/scala/org/apache/livy/utils/SparkYarnApp.scala:
##########
@@ -82,7 +94,13 @@ object SparkYarnApp extends Logging {
// kill the app if found it and remove it if exceeding a threshold
val iter = leakedAppTags.entrySet().iterator()
val now = System.currentTimeMillis()
- val apps = client.getApplications(appType).asScala
+ val tagSet = new util.HashSet[String]()
Review Comment:
The second iterator (entries) only exists to copy keys into a HashSet. This can be simplified to:
```suggestion
val tagSet = new util.HashSet[String]()
val tagSet = new util.HashSet[String](leakedAppTags.keySet())
```
This is one line instead of four does the same thing and eliminates the redundant iterator.
##########
server/src/test/scala/org/apache/livy/utils/SparkYarnAppSpec.scala:
##########
@@ -49,6 +50,13 @@ class SparkYarnAppSpec extends FunSpec with LivyBaseUnitTestSuite {
Thread.`yield`()
}
+ private def mockGetApplicationsByTags(
+ client: YarnClient,
+ reports: List[ApplicationReport]): Unit = {
+ when(client.getApplications(any(classOf[GetApplicationsRequest])))
Review Comment:
Using `any(classOf[GetApplicationsRequest])` is a bit broad, it won't catch bugs if wrong tags or app types are passed later. How about using an ArgumentCaptor here to verify the exact fields? For example:
```
diff --git a/server/src/test/scala/org/apache/livy/utils/SparkYarnAppSpec.scala b/server/src/test/scala/org/apache/livy/utils/SparkYarnAppSpec.scala
index cfaca49..55d69c4 100644
--- a/server/src/test/scala/org/apache/livy/utils/SparkYarnAppSpec.scala
+++ b/server/src/test/scala/org/apache/livy/utils/SparkYarnAppSpec.scala
@@ -30,6 +30,7 @@ import org.apache.hadoop.yarn.api.records.YarnApplicationState._
import org.apache.hadoop.yarn.client.api.YarnClient
import org.apache.hadoop.yarn.exceptions.ApplicationAttemptNotFoundException
import org.apache.hadoop.yarn.util.ConverterUtils
+import org.mockito.ArgumentCaptor
import org.mockito.Matchers.any
import org.mockito.Mockito._
import org.mockito.invocation.InvocationOnMock
@@ -411,6 +412,14 @@ class SparkYarnAppSpec extends FunSpec with LivyBaseUnitTestSuite {
verify(mockYarnClient, atLeast(1)).getApplicationReport(appId)
verify(mockListener).appIdKnown(appId.toString)
+
+ val requestCaptor = ArgumentCaptor.forClass(classOf[GetApplicationsRequest])
+ verify(mockYarnClient, atLeast(1)).getApplications(requestCaptor.capture())
+ val capturedRequest = requestCaptor.getValue
+ assert(capturedRequest.getApplicationTags.contains(appTag.toLowerCase),
+ s"Request must contain the lowercase appTag '${appTag.toLowerCase}'")
+ assert(capturedRequest.getApplicationTypes.contains("SPARK"),
+ "Request must filter by application type 'SPARK'")
}
}
}
@@ -685,11 +694,20 @@ class SparkYarnAppSpec extends FunSpec with LivyBaseUnitTestSuite {
SparkYarnApp.init(livyConf, Some(client))
SparkYarnApp.leakedAppTags.clear()
- SparkYarnApp.leakedAppTags.put("leakApp", System.currentTimeMillis())
+ val leakAppTag = "leakApp"
+ SparkYarnApp.leakedAppTags.put(leakAppTag, System.currentTimeMillis())
Eventually.eventually(Eventually.timeout(TEST_TIMEOUT), Eventually.interval(100 millis)) {
assert(SparkYarnApp.leakedAppTags.size() == 0)
}
+
+ val requestCaptor = ArgumentCaptor.forClass(classOf[GetApplicationsRequest])
+ verify(client, atLeast(1)).getApplications(requestCaptor.capture())
+ val capturedRequest = requestCaptor.getValue
+ assert(capturedRequest.getApplicationTags.contains(leakAppTag.toLowerCase),
+ "Leaked app GC request must include the leaked app tag")
+ assert(capturedRequest.getApplicationTypes.contains("SPARK"),
+ "Leaked app GC request must filter by application type 'SPARK'")
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@livy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] LIVY-866: Optimize Yarn GetApplications Query to prevent additional load on Yarn and Livy [livy]
Posted by "roczei (via GitHub)" <gi...@apache.org>.
roczei commented on code in PR #533:
URL: https://github.com/apache/livy/pull/533#discussion_r3664898366
##########
server/src/main/scala/org/apache/livy/utils/SparkYarnApp.scala:
##########
@@ -35,13 +38,22 @@ import org.apache.livy.{LivyConf, Logging, Utils}
object SparkYarnApp extends Logging {
+ @volatile private var leakedAppsGCThreadStarted = false
+
def init(livyConf: LivyConf, client: Option[YarnClient] = None): Unit = {
mockYarnClient = client
sessionLeakageCheckInterval = livyConf.getTimeAsMs(LivyConf.YARN_APP_LEAKAGE_CHECK_INTERVAL)
sessionLeakageCheckTimeout = livyConf.getTimeAsMs(LivyConf.YARN_APP_LEAKAGE_CHECK_TIMEOUT)
- leakedAppsGCThread.setDaemon(true)
- leakedAppsGCThread.setName("LeakedAppsGCThread")
- leakedAppsGCThread.start()
+ if (!leakedAppsGCThreadStarted) {
+ synchronized {
Review Comment:
@nileshrathi345
Why do we need this synchronized block? Could we avoid it and restore the previous behavior?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@livy.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org