[SPARK-31235][FOLLOWUP][TESTS][TEST-HADOOP3.2] Fix test "specify a more specific type for the ap…
### What changes were proposed in this pull request? Update the input parameters for instantiating `RMAppManager` and `ClientRMService` ### Why are the changes needed? For hadoop3.2, if `RMAppManager` is not created correctly, the following exception will occur: ``` java.lang.RuntimeException: java.lang.NullPointerException at org.apache.hadoop.util.ReflectionUtils.newInstance(ReflectionUtils.java:135) at org.apache.hadoop.yarn.security.YarnAuthorizationProvider.getInstance(YarnAuthorizationProvider.java:55) at org.apache.hadoop.yarn.server.resourcemanager.RMAppManager.<init>(RMAppManager.java:117) ``` ### How was this patch tested? UTs Closes #28456 from wang-zhun/Fix-SPARK-31235. Authored-by: wang-zhun <wangzhun6103@gmail.com> Signed-off-by: Thomas Graves <tgraves@apache.org>
This commit is contained in:
parent
77c690a725
commit
c1801fd6da
|
@ -38,6 +38,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.{ClientRMService, RMAppMana
|
||||||
import org.apache.hadoop.yarn.server.resourcemanager.ahs.RMApplicationHistoryWriter
|
import org.apache.hadoop.yarn.server.resourcemanager.ahs.RMApplicationHistoryWriter
|
||||||
import org.apache.hadoop.yarn.server.resourcemanager.metrics.SystemMetricsPublisher
|
import org.apache.hadoop.yarn.server.resourcemanager.metrics.SystemMetricsPublisher
|
||||||
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp
|
import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp
|
||||||
|
import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler
|
||||||
import org.apache.hadoop.yarn.server.security.ApplicationACLsManager
|
import org.apache.hadoop.yarn.server.security.ApplicationACLsManager
|
||||||
import org.apache.hadoop.yarn.util.Records
|
import org.apache.hadoop.yarn.util.Records
|
||||||
import org.mockito.ArgumentMatchers.{any, anyBoolean, anyShort, eq => meq}
|
import org.mockito.ArgumentMatchers.{any, anyBoolean, anyShort, eq => meq}
|
||||||
|
@ -222,32 +223,11 @@ class ClientSuite extends SparkFunSuite with Matchers {
|
||||||
3 -> ("SPARK-SQL", "SPARK-SQL"),
|
3 -> ("SPARK-SQL", "SPARK-SQL"),
|
||||||
4 -> ("012345678901234567890123", "01234567890123456789"))
|
4 -> ("012345678901234567890123", "01234567890123456789"))
|
||||||
|
|
||||||
for ((id, (sourceType, targetType)) <- appTypes) {
|
// Mock yarn submit application
|
||||||
val sparkConf = new SparkConf().set("spark.yarn.applicationType", sourceType)
|
|
||||||
val args = new ClientArguments(Array())
|
|
||||||
|
|
||||||
val appContext = spy(Records.newRecord(classOf[ApplicationSubmissionContext]))
|
|
||||||
val appId = ApplicationId.newInstance(123456, id)
|
|
||||||
appContext.setApplicationId(appId)
|
|
||||||
val getNewApplicationResponse = Records.newRecord(classOf[GetNewApplicationResponse])
|
|
||||||
val containerLaunchContext = Records.newRecord(classOf[ContainerLaunchContext])
|
|
||||||
|
|
||||||
val client = new Client(args, sparkConf, null)
|
|
||||||
val context = client.createApplicationSubmissionContext(
|
|
||||||
new YarnClientApplication(getNewApplicationResponse, appContext),
|
|
||||||
containerLaunchContext)
|
|
||||||
|
|
||||||
val yarnClient = mock(classOf[YarnClient])
|
val yarnClient = mock(classOf[YarnClient])
|
||||||
when(yarnClient.submitApplication(any())).thenAnswer((invocationOnMock: InvocationOnMock) => {
|
val rmApps = new ConcurrentHashMap[ApplicationId, RMApp]()
|
||||||
val subContext = invocationOnMock.getArguments()(0)
|
|
||||||
.asInstanceOf[ApplicationSubmissionContext]
|
|
||||||
val request = Records.newRecord(classOf[SubmitApplicationRequest])
|
|
||||||
request.setApplicationSubmissionContext(subContext)
|
|
||||||
|
|
||||||
val rmContext = mock(classOf[RMContext])
|
val rmContext = mock(classOf[RMContext])
|
||||||
val conf = mock(classOf[Configuration])
|
when(rmContext.getRMApps).thenReturn(rmApps)
|
||||||
val map = new ConcurrentHashMap[ApplicationId, RMApp]()
|
|
||||||
when(rmContext.getRMApps).thenReturn(map)
|
|
||||||
val dispatcher = mock(classOf[Dispatcher])
|
val dispatcher = mock(classOf[Dispatcher])
|
||||||
when(rmContext.getDispatcher).thenReturn(dispatcher)
|
when(rmContext.getDispatcher).thenReturn(dispatcher)
|
||||||
when[EventHandler[_]](dispatcher.getEventHandler).thenReturn(
|
when[EventHandler[_]](dispatcher.getEventHandler).thenReturn(
|
||||||
|
@ -259,26 +239,46 @@ class ClientSuite extends SparkFunSuite with Matchers {
|
||||||
when(rmContext.getRMApplicationHistoryWriter).thenReturn(writer)
|
when(rmContext.getRMApplicationHistoryWriter).thenReturn(writer)
|
||||||
val publisher = mock(classOf[SystemMetricsPublisher])
|
val publisher = mock(classOf[SystemMetricsPublisher])
|
||||||
when(rmContext.getSystemMetricsPublisher).thenReturn(publisher)
|
when(rmContext.getSystemMetricsPublisher).thenReturn(publisher)
|
||||||
when(appContext.getUnmanagedAM).thenReturn(true)
|
val yarnScheduler = mock(classOf[YarnScheduler])
|
||||||
|
|
||||||
val rmAppManager = new RMAppManager(rmContext,
|
val rmAppManager = new RMAppManager(rmContext,
|
||||||
null,
|
yarnScheduler,
|
||||||
null,
|
null,
|
||||||
mock(classOf[ApplicationACLsManager]),
|
mock(classOf[ApplicationACLsManager]),
|
||||||
conf)
|
new Configuration())
|
||||||
val clientRMService = new ClientRMService(rmContext,
|
val clientRMService = new ClientRMService(rmContext,
|
||||||
null,
|
yarnScheduler,
|
||||||
rmAppManager,
|
rmAppManager,
|
||||||
null,
|
null,
|
||||||
null,
|
null,
|
||||||
null)
|
null)
|
||||||
|
clientRMService.init(new Configuration())
|
||||||
|
when(yarnClient.submitApplication(any())).thenAnswer((invocationOnMock: InvocationOnMock) => {
|
||||||
|
val subContext = invocationOnMock.getArguments()(0)
|
||||||
|
.asInstanceOf[ApplicationSubmissionContext]
|
||||||
|
val request = Records.newRecord(classOf[SubmitApplicationRequest])
|
||||||
|
request.setApplicationSubmissionContext(subContext)
|
||||||
clientRMService.submitApplication(request)
|
clientRMService.submitApplication(request)
|
||||||
|
|
||||||
assert(map.get(subContext.getApplicationId).getApplicationType === targetType)
|
|
||||||
null
|
null
|
||||||
})
|
})
|
||||||
|
|
||||||
|
// Spark submit application
|
||||||
|
val appContext = spy(Records.newRecord(classOf[ApplicationSubmissionContext]))
|
||||||
|
when(appContext.getUnmanagedAM).thenReturn(true)
|
||||||
|
for ((id, (sourceType, targetType)) <- appTypes) {
|
||||||
|
val sparkConf = new SparkConf().set("spark.yarn.applicationType", sourceType)
|
||||||
|
val args = new ClientArguments(Array())
|
||||||
|
val appId = ApplicationId.newInstance(123456, id)
|
||||||
|
appContext.setApplicationId(appId)
|
||||||
|
val getNewApplicationResponse = Records.newRecord(classOf[GetNewApplicationResponse])
|
||||||
|
val containerLaunchContext = Records.newRecord(classOf[ContainerLaunchContext])
|
||||||
|
|
||||||
|
val client = new Client(args, sparkConf, null)
|
||||||
|
val context = client.createApplicationSubmissionContext(
|
||||||
|
new YarnClientApplication(getNewApplicationResponse, appContext),
|
||||||
|
containerLaunchContext)
|
||||||
|
|
||||||
yarnClient.submitApplication(context)
|
yarnClient.submitApplication(context)
|
||||||
|
assert(rmApps.get(appId).getApplicationType === targetType)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue