resolve k8s conflicts

Author: Robert Kruszewski

common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/kubernetes/KubernetesExternalShuffleClient.java

@@ -47,8 +47,9 @@ public class KubernetesExternalShuffleClient extends ExternalShuffleClient {
   public KubernetesExternalShuffleClient(
       TransportConf conf,
       SecretKeyHolder secretKeyHolder,
-      boolean authEnabled) {
-    super(conf, secretKeyHolder, authEnabled);
+      boolean authEnabled,
+      long registrationTimeoutMs) {
+    super(conf, secretKeyHolder, authEnabled, registrationTimeoutMs);
   }
 
   public void registerDriverWithShuffleService(String host, int port)

resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/kubernetes/KubernetesClusterSchedulerBackend.scala

@@ -35,6 +35,7 @@ import org.apache.spark.{SparkContext, SparkEnv, SparkException}
 import org.apache.spark.deploy.kubernetes.{ConfigurationUtils, SparkPodInitContainerBootstrap}
 import org.apache.spark.deploy.kubernetes.config._
 import org.apache.spark.deploy.kubernetes.constants._
+import org.apache.spark.internal.config
 import org.apache.spark.network.netty.SparkTransportConf
 import org.apache.spark.network.shuffle.kubernetes.KubernetesExternalShuffleClient
 import org.apache.spark.rpc.{RpcCallContext, RpcEndpointAddress, RpcEnv}
@@ -209,7 +210,8 @@ private[spark] class KubernetesClusterSchedulerBackend(
     new KubernetesExternalShuffleClient(
       SparkTransportConf.fromSparkConf(conf, "shuffle"),
       sc.env.securityManager,
-      sc.env.securityManager.isAuthenticationEnabled())
+      sc.env.securityManager.isAuthenticationEnabled(),
+      conf.get(config.SHUFFLE_REGISTRATION_TIMEOUT))
   }
 
   private def getInitialTargetExecutorNumber(defaultNumExecutors: Int = 1): Int = {

resource-managers/kubernetes/core/src/test/scala/org/apache/spark/deploy/kubernetes/submit/KubernetesExternalShuffleServiceSuite.scala

@@ -19,6 +19,7 @@ package org.apache.spark.deploy.kubernetes.submit
 import org.apache.spark.{SecurityManager, SparkConf, SparkFunSuite}
 import org.apache.spark.deploy.kubernetes.config._
 import org.apache.spark.deploy.kubernetes.KubernetesExternalShuffleService
+import org.apache.spark.internal.config
 import org.apache.spark.network.netty.SparkTransportConf
 import org.apache.spark.network.shuffle.kubernetes.KubernetesExternalShuffleClient
 
@@ -35,7 +36,8 @@ private[spark] class KubernetesExternalShuffleServiceSuite extends SparkFunSuite
     val shuffleClient = new KubernetesExternalShuffleClient(
       SparkTransportConf.fromSparkConf(SPARK_CONF, "shuffle"),
       new SecurityManager(SPARK_CONF),
-      false)
+      false,
+      SPARK_CONF.get(config.SHUFFLE_REGISTRATION_TIMEOUT))
 
     shuffleService.start()
     shuffleClient.init("newapp")