protected SparkAppDriverConf buildDriverConf(
      SparkApplication app, Map<String, String> confOverrides) {
    ApplicationSpec applicationSpec = app.getSpec();
    SparkConf effectiveSparkConf = new SparkConf();
    if (MapUtils.isNotEmpty(applicationSpec.getSparkConf())) {
      for (String confKey : applicationSpec.getSparkConf().keySet()) {
        effectiveSparkConf.set(confKey, applicationSpec.getSparkConf().get(confKey));
      }
    }
    if (MapUtils.isNotEmpty(confOverrides)) {
      for (Map.Entry<String, String> entry : confOverrides.entrySet()) {
        effectiveSparkConf.set(entry.getKey(), entry.getValue());
      }
    }
    effectiveSparkConf.set("spark.kubernetes.namespace", app.getMetadata().getNamespace());
    MainAppResource primaryResource = new JavaMainAppResource(Option.empty());
    if (StringUtils.isNotEmpty(applicationSpec.getJars())) {
      primaryResource = new JavaMainAppResource(Option.apply(applicationSpec.getJars()));
      effectiveSparkConf.setIfMissing("spark.jars", applicationSpec.getJars());
    } else if (StringUtils.isNotEmpty(applicationSpec.getPyFiles())) {
      primaryResource = new PythonMainAppResource(applicationSpec.getPyFiles());
      effectiveSparkConf.setIfMissing("spark.submit.pyFiles", applicationSpec.getPyFiles());
    } else if (StringUtils.isNotEmpty(applicationSpec.getSparkRFiles())) {
      primaryResource = new RMainAppResource(applicationSpec.getSparkRFiles());
    }
    String sparkMasterUrlPrefix =
        effectiveSparkConf.get(MASTER_URL_PREFIX_PROPS_NAME, DEFAULT_MASTER_URL_PREFIX);
    effectiveSparkConf.setIfMissing(
        "spark.master",
        sparkMasterUrlPrefix + "https://$KUBERNETES_SERVICE_HOST:$KUBERNETES_SERVICE_PORT");
    String appId = generateSparkAppId(app);
    effectiveSparkConf.setIfMissing("spark.app.id", appId);
    return SparkAppDriverConf.create(
        effectiveSparkConf,
        appId,
        primaryResource,
        applicationSpec.getMainClass(),
        applicationSpec.getDriverArgs().toArray(String[]::new),
        Option.apply(applicationSpec.getProxyUser()));
  }
@Test
  void buildDriverConfShouldApplySpecAndPropertiesOverride() {
    Map<SparkAppDriverConf, List<Object>> constructorArgs = new HashMap<>();
    try (MockedConstruction<SparkAppDriverConf> mocked =
        mockConstruction(
            SparkAppDriverConf.class,
            (mock, context) -> constructorArgs.put(mock, new ArrayList<>(context.arguments())))) {
      SparkApplication mockApp = mock(SparkApplication.class);
      ApplicationSpec mockSpec = mock(ApplicationSpec.class);
      ObjectMeta appMeta = new ObjectMetaBuilder().withName("app1").withNamespace("ns1").build();
      Map<String, String> appProps = new HashMap<>();
      appProps.put("foo", "bar");
      appProps.put("spark.executor.instances", "1");
      appProps.put("spark.kubernetes.namespace", "ns2");
      Map<String, String> overrides = new HashMap<>();
      overrides.put("spark.executor.instances", "5");
      overrides.put("spark.kubernetes.namespace", "ns3");
      when(mockSpec.getSparkConf()).thenReturn(appProps);
      when(mockApp.getSpec()).thenReturn(mockSpec);
      when(mockApp.getMetadata()).thenReturn(appMeta);
      when(mockSpec.getProxyUser()).thenReturn("foo-user");
      when(mockSpec.getMainClass()).thenReturn("foo-class");
      when(mockSpec.getDriverArgs()).thenReturn(List.of("a", "b"));

      SparkAppSubmissionWorker submissionWorker = new SparkAppSubmissionWorker();
      SparkAppDriverConf conf = submissionWorker.buildDriverConf(mockApp, overrides);
      assertEquals(6, constructorArgs.get(conf).size());

      // validate SparkConf with override
      assertTrue(constructorArgs.get(conf).get(0) instanceof SparkConf);
      SparkConf createdConf = (SparkConf) constructorArgs.get(conf).get(0);
      assertEquals("bar", createdConf.get("foo"));
      assertEquals("5", createdConf.get("spark.executor.instances"));

      assertEquals(
          "ns1",
          createdConf.get("spark.kubernetes.namespace"),
          "namespace from CR takes highest precedence");

      // validate main resources
      assertTrue(constructorArgs.get(conf).get(2) instanceof JavaMainAppResource);
      JavaMainAppResource mainResource = (JavaMainAppResource) constructorArgs.get(conf).get(2);
      assertTrue(mainResource.primaryResource().isEmpty());

      assertEquals("foo-class", constructorArgs.get(conf).get(3));

      assertTrue(constructorArgs.get(conf).get(4) instanceof String[]);
      String[] capturedArgs = (String[]) constructorArgs.get(conf).get(4);
      assertEquals(2, capturedArgs.length);
      assertEquals("a", capturedArgs[0]);
      assertEquals("b", capturedArgs[1]);
    }
  }