Avoid "Manually specified and authenticated users mismatch" YT error"

spark-submit/src/main/scala/tech/ytsaurus/spyt/submit/SparkCluster.scala

@@ -15,8 +15,8 @@ case class SparkCluster(master: String,
                         masterHostAndPort: HostAndPort)
 
 object SparkCluster {
-  def get(proxy: String, baseDiscoveryPath: String, user: String, token: String): SparkCluster = {
-    val ytClient = YtWrapper.createRpcClient("submission client", YtClientConfiguration.default(proxy, user, token))
+  def get(proxy: String, baseDiscoveryPath: String, token: String): SparkCluster = {
+    val ytClient = YtWrapper.createRpcClient("submission client", YtClientConfiguration.default(proxy, token))
     try {
       implicit val yt = ytClient.yt
       val discoveryService = new CypressDiscoveryService(baseDiscoveryPath)

spark-submit/src/main/scala/tech/ytsaurus/spyt/submit/SubmissionClient.scala

@@ -21,13 +21,12 @@ import scala.util.{Failure, Success, Try}
 
 class SubmissionClient(proxy: String,
                        baseDiscoveryPath: String,
-                       user: String,
                        token: String) {
   private val log = LoggerFactory.getLogger(getClass)
 
   private val eventLogPath = CypressDiscoveryService.eventLogPath(baseDiscoveryPath)
 
-  private val cluster = new AtomicReference[SparkCluster](SparkCluster.get(proxy, baseDiscoveryPath, user, token))
+  private val cluster = new AtomicReference[SparkCluster](SparkCluster.get(proxy, baseDiscoveryPath, token))
 
   private val threadFactory = new ThreadFactory() {
     override def newThread(runnable: Runnable): Thread = {
@@ -51,7 +50,7 @@ class SubmissionClient(proxy: String,
 
   def submit(launcher: InProcessLauncher,
              retryConfig: RetryConfig): Try[String] = {
-    val yt = YtClientProvider.ytClient(YtClientConfiguration.default(proxy, user, token))
+    val yt = YtClientProvider.ytClient(YtClientConfiguration.default(proxy, token))
     val remoteGlobalConfig = parseRemoteConfig(remoteGlobalConfigPath, yt)
     val remoteVersionConfig = parseRemoteConfig(remoteVersionConfigPath(cluster.get().version), yt)
     val remoteClusterConfig = parseRemoteConfig(remoteClusterConfigPath(baseDiscoveryPath), yt)
@@ -187,7 +186,7 @@ class SubmissionClient(proxy: String,
 
   private def updateCluster(): Unit = {
     log.debug(s"Update cluster addresses from $baseDiscoveryPath")
-    cluster.set(SparkCluster.get(proxy, baseDiscoveryPath, user, token))
+    cluster.set(SparkCluster.get(proxy, baseDiscoveryPath, token))
   }
 
   private def forceClusterUpdate(): Unit = {

yt-wrapper/src/main/scala/tech/ytsaurus/spyt/wrapper/client/DefaultRpcCredentials.scala

@@ -1,19 +1,12 @@
 package tech.ytsaurus.spyt.wrapper.client
 
 import java.nio.file.{Files, Path, Paths}
-import tech.ytsaurus.client.rpc.YTsaurusClientAuth
 
 object DefaultRpcCredentials {
   def token: String = {
     sys.env.getOrElse("YT_TOKEN", readFileFromHome(".yt", "token"))
   }
 
-  def user: String = sys.env.getOrElse("YT_USER", System.getProperty("user.name"))
-
-  def credentials: YTsaurusClientAuth = {
-    YTsaurusClientAuth.builder().setUser(user).setToken(token).build()
-  }
-
   private def readFileFromHome(path: String*): String = {
     readFile(Paths.get(System.getProperty("user.home"), path: _*))
   }

yt-wrapper/src/main/scala/tech/ytsaurus/spyt/wrapper/client/YtClientConfiguration.scala

@@ -14,7 +14,6 @@ import scala.util.{Failure, Success, Try}
 
 @SerialVersionUID(6764302982752098915L)
 case class YtClientConfiguration(proxy: String,
-                                 user: String,
                                  token: String,
                                  timeout: Duration,
                                  proxyRole: Option[String],
@@ -41,7 +40,7 @@ case class YtClientConfiguration(proxy: String,
 
   def isHttps: Boolean = proxyUrl.toOption.exists(_.getProtocol == "https")
 
-  def clientAuth: YTsaurusClientAuth = YTsaurusClientAuth.builder().setUser(user).setToken(token).build()
+  def clientAuth: YTsaurusClientAuth = YTsaurusClientAuth.builder().setToken(token).build()
 
   def replaceProxy(newProxy: Option[String]): YtClientConfiguration = {
     if (newProxy.isDefined && newProxy.get != proxy) {
@@ -71,7 +70,6 @@ object YtClientConfiguration {
       getByName("proxy").orElse(sys.env.get("YT_PROXY")).getOrElse(
         throw new IllegalArgumentException("Proxy must be specified")
       ),
-      getByName("user").orElse(sys.env.get("YT_SECURE_VAULT_YT_USER")).getOrElse(DefaultRpcCredentials.user),
       getByName("token").orElse(sys.env.get("YT_SECURE_VAULT_YT_TOKEN")).getOrElse(DefaultRpcCredentials.token),
       getByName("timeout").map(Utils.parseDuration).getOrElse(60 seconds),
       getByName("proxyRole"),
@@ -104,13 +102,11 @@ object YtClientConfiguration {
 
   def default(proxy: String): YtClientConfiguration = default(
     proxy = proxy,
-    user = DefaultRpcCredentials.user,
     token = DefaultRpcCredentials.token
   )
 
-  def default(proxy: String, user: String, token: String): YtClientConfiguration = YtClientConfiguration(
+  def default(proxy: String, token: String): YtClientConfiguration = YtClientConfiguration(
     proxy = proxy,
-    user = user,
     token = token,
     timeout = 5 minutes,
     proxyRole = None,
@@ -121,15 +117,14 @@ object YtClientConfiguration {
   )
 
   def create(proxy: String,
-             user: String,
              token: String,
              timeout: JDuration,
              proxyRole: String,
              byop: ByopConfiguration,
              masterWrapperUrl: String,
              extendedFileTimeout: Boolean,
              proxyNetworkName: Option[String] = None) = new YtClientConfiguration(
-    proxy, user, token, toScalaDuration(timeout),
+    proxy, token, toScalaDuration(timeout),
     Option(proxyRole), byop, Option(masterWrapperUrl), extendedFileTimeout, proxyNetworkName
   )
 }

yt-wrapper/src/test/scala/tech/ytsaurus/spyt/test/LocalYtClient.scala

@@ -31,7 +31,6 @@ trait LocalYtClient extends LocalYt {
 
   private val conf: YtClientConfiguration = YtClientConfiguration(
     proxy = LocalYt.proxy,
-    user = "root",
     token = "",
     timeout = 5 minutes,
     proxyRole = None,