Issue while reading hdfs file post Kerberos Authn

[Susmit07]

Hi Developers

We are facing an issue when using Apache Pekko connectors wrapped within a UserGroupInformation.doAs block for Kerberos authentication. The same HDFS operation works fine in a single-threaded Scala program without Pekko connectors, but fails when run in Pekko streams. Are there any known limitations or configurations needed for Pekko connectors when dealing with Kerberos-authenticated HDFS access or concurrent UGI contexts?

def main(args: Array[String]): Unit = {

    val hdfsPath = "hdfs://<dir=path>"
    println("Path>>" + hdfsPath)

    val hdfsConf = KerberosHadoopAuthenticator.hadoopConfig

    val ugi: UserGroupInformation = KerberosHadoopAuthenticator.getAuthenticatedUGI(None, hdfsConf)

    ugi.doAs(new PrivilegedExceptionAction[Unit] {
      override def run(): Unit = {
        println("Inside run >>")
        val fs = FileSystem.get(hdfsConf)
        println("fs initialized >>")
        val files = fs.listStatus(new Path(hdfsPath))
        files.foreach(fileStatus => println(fileStatus.getPath.toString))
        fs.close()
      }
    })
  }

The above code runs fine

Inside run >>
[2024-10-07 19:58:41,503] [WARN] [org.apache.hadoop.hdfs.shortcircuit.DomainSocketFactory] [] [main] - The short-circuit local reads feature cannot be used because libhadoop cannot be loaded. MDC: {}
fs initialized >>
hdfs://nnproxies/user/svc-segscores-dev/cef-nxt/sample.parquet

Now when accessing using pekko streams

override def listFiles(sourceDirectory: String): Future[List[String]] = {
    ugi.doAs(new PrivilegedExceptionAction[Future[List[String]]] {
      override def run(): Future[List[String]] = listParquetFilesInDirectory(sourceDirectory)
    })
  }

private def listParquetFilesInDirectory(directory: String): Future[List[String]] = Future {
    try {
      val status = fs.listStatus(new HadoopPath(directory)).toList
      status.flatMap { fileStatus =>
        if (fileStatus.isDirectory) {
          listParquetFilesInDirectory(fileStatus.getPath.toString).value.get.get
        } else if (fileStatus.isFile && fileStatus.getPath.toString.toLowerCase.endsWith(".parquet")) {
          Some(fileStatus.getPath.toString)
        } else {
          None
        }
      }
    } catch {
      case _: FileNotFoundException =>
        logger.error(s"Directory not found: $directory")
        List.empty[String]
      case ex: Exception =>
        logger.error(s"Error accessing directory: ${ex.getMessage}")
        List.empty[String]
    }
  }

The above function fails with below error:

[2024-10-07 17:22:02,403] [WARN] [org.apache.hadoop.ipc.Client] [] [cef-datasync-downloader-pekko.actor.default-dispatcher-21] - Exception encountered while connecting to the server : org.apache.hadoop.security.AccessControlException: Client cannot authenticate via:[TOKEN, KERBEROS] MDC: {}
[2024-10-07 17:22:02,418] [WARN] [org.apache.hadoop.ipc.Client] [] [cef-datasync-downloader-pekko.actor.default-dispatcher-21] - Exception encountered while connecting to the server : org.apache.hadoop.security.AccessControlException: Client cannot authenticate via:[TOKEN, KERBEROS] MDC: {}
[2024-10-07 17:22:02,418] [INFO] [org.apache.hadoop.io.retry.RetryInvocationHandler] [] [cef-datasync-downloader-pekko.actor.default-dispatcher-21] - java.io.IOException: DestHost:destPort hdd0935.global.tesco.org:8890 , LocalHost:localPort cef-datasync-downloader-575b5d755c-57mjc/10.42.2.113:0. Failed on local exception: java.io.IOException: org.apache.hadoop.security.AccessControlException: Client cannot authenticate via:[TOKEN, KERBEROS], while invoking ClientNamenodeProtocolTranslatorPB.getListing over hdd0935.global.tesco.org/172.17.121.235:8890 after 1 failover attempts. Trying to failover after sleeping for 650ms. MDC: {}

Requesting for guidance here - @pjfanning

[Susmit07]

@raboof sorry I am tagging but we are kinda blocked, not sure if it's a limitation which shouldn't be

[pjfanning]

Futures run in a different thread so setting user permissions before delegating to a Future that runs asynchronously in another thread doesn't seem like a good idea.

[Susmit07]

you are correct

Since Futures run on a separate thread pool, wrapping the Future itself inside a ugi.doAs block does not guarantee that the UGI credentials will propagate to the thread executing the Future. Instead, we can run the entire Future logic synchronously within the ugi.doAs block, and then wrap the result in a Future afterwards, which worked

[Susmit07]

Thank you for the million dollar suggestion

[2024-10-07 20:55:15,779] [INFO] [com.tesco.cef.datasync.processor.hdfs.HdfsSyncProcessor] [] [cef-datasync-downloader-pekko.actor.default-dispatcher-13] - Processed record from hdfs://nnproxies/user/svc-segscores-dev/cef-nxt/sample.parquet: {"column0": "Leo", "column1": "Walters"} MDC: {}
[2024-10-07 20:55:15,779] [INFO] [com.tesco.cef.datasync.processor.hdfs.HdfsSyncProcessor] [] [cef-datasync-downloader-pekko.actor.default-dispatcher-13] - Processed record from hdfs://nnproxies/user/svc-segscores-dev/cef-nxt/sample.parquet: {"column0": "Samuel", "column1": "Schmidt"} MDC: {}
[2024-10-07 20:55:15,779] [INFO] [com.tesco.cef.datasync.processor.hdfs.HdfsSyncProcessor] [] [cef-datasync-downloader-pekko.actor.default-dispatcher-13] - Processed record from hdfs://nnproxies/user/svc-segscores-dev/cef-nxt/sample.parquet: {"column0": "Elizabeth", "column1": "Gordon"} MDC: {}