You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@pekko.apache.org by "Susmit07 (via GitHub)" <gi...@apache.org> on 2024/10/07 20:08:34 UTC
[D] Issue while reading hdfs file post Kerberos Authn [pekko-connectors]
GitHub user Susmit07 created a discussion: Issue while reading hdfs file post Kerberos Authn
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
GitHub link: https://github.com/apache/pekko-connectors/discussions/858
----
This is an automatically sent email for notifications@pekko.apache.org.
To unsubscribe, please send an email to: notifications-unsubscribe@pekko.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [D] Issue while reading hdfs file post Kerberos Authn [pekko-connectors]
Posted by "Susmit07 (via GitHub)" <gi...@apache.org>.
GitHub user Susmit07 closed the discussion with a comment: Issue while reading hdfs file post Kerberos Authn
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
GitHub link: https://github.com/apache/pekko-connectors/discussions/858#discussioncomment-10873090
----
This is an automatically sent email for notifications@pekko.apache.org.
To unsubscribe, please send an email to: notifications-unsubscribe@pekko.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [D] Issue while reading hdfs file post Kerberos Authn [pekko-connectors]
Posted by "Susmit07 (via GitHub)" <gi...@apache.org>.
GitHub user Susmit07 closed the discussion with a comment: Issue while reading hdfs file post Kerberos Authn
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: {}
```
GitHub link: https://github.com/apache/pekko-connectors/discussions/858#discussioncomment-10873061
----
This is an automatically sent email for notifications@pekko.apache.org.
To unsubscribe, please send an email to: notifications-unsubscribe@pekko.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [D] Issue while reading hdfs file post Kerberos Authn [pekko-connectors]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
GitHub user pjfanning added a comment to the discussion: Issue while reading hdfs file post Kerberos Authn
Futures don't 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.
GitHub link: https://github.com/apache/pekko-connectors/discussions/858#discussioncomment-10872781
----
This is an automatically sent email for notifications@pekko.apache.org.
To unsubscribe, please send an email to: notifications-unsubscribe@pekko.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [D] Issue while reading hdfs file post Kerberos Authn [pekko-connectors]
Posted by "pjfanning (via GitHub)" <gi...@apache.org>.
GitHub user pjfanning edited a comment on the discussion: Issue while reading hdfs file post Kerberos Authn
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.
GitHub link: https://github.com/apache/pekko-connectors/discussions/858#discussioncomment-10872781
----
This is an automatically sent email for notifications@pekko.apache.org.
To unsubscribe, please send an email to: notifications-unsubscribe@pekko.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [D] Issue while reading hdfs file post Kerberos Authn [pekko-connectors]
Posted by "Susmit07 (via GitHub)" <gi...@apache.org>.
GitHub user Susmit07 added a comment to the discussion: Issue while reading hdfs file post Kerberos Authn
@raboof sorry I am tagging but we are kinda blocked, not sure if it's a limitation which shouldn't be
GitHub link: https://github.com/apache/pekko-connectors/discussions/858#discussioncomment-10872733
----
This is an automatically sent email for notifications@pekko.apache.org.
To unsubscribe, please send an email to: notifications-unsubscribe@pekko.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org
Re: [D] Issue while reading hdfs file post Kerberos Authn [pekko-connectors]
Posted by "Susmit07 (via GitHub)" <gi...@apache.org>.
GitHub user Susmit07 closed a discussion: Issue while reading hdfs file post Kerberos Authn
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
GitHub link: https://github.com/apache/pekko-connectors/discussions/858
----
This is an automatically sent email for notifications@pekko.apache.org.
To unsubscribe, please send an email to: notifications-unsubscribe@pekko.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org