yuqi1129 commented on code in PR #13455:
URL: https://github.com/apache/gravitino/pull/13455#discussion_r4231032617
##########
clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/BaseGVFSOperations.java:
##########
@@ -453,23 +455,72 @@ public abstract FSDataOutputStream append(Path gvfsPath,
int bufferSize, Progres
*/
public abstract Token<?>[] addDelegationTokens(String renewer, Credentials
credentials);
+ /**
+ * Resolve the filesets listed in {@link
+ *
GravitinoVirtualFileSystemConfiguration#FS_GRAVITINO_DELEGATION_TOKEN_FILESETS},
so that their
+ * actual file systems exist before delegation tokens are collected. Token
requests carry no path,
+ * so the filesets cannot be learned from the caller.
+ */
+ public void prepareDelegationTokens() {
+ for (String path :
+ conf.getTrimmedStrings(
+
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_DELEGATION_TOKEN_FILESETS))
{
+ try {
+ getActualFileSystem(new Path(path), currentLocationName());
+ } catch (Exception e) {
+ // Never fail token collection: an unresolvable fileset simply yields
no token.
+ LOG.warn("Failed to resolve {} while preparing delegation tokens",
path, e);
+ }
+ }
+ }
+
/**
* Add delegation tokens for all file systems in the cache.
*
+ * <p>Only tokens owned by the current user are added. A fileset configured
with its own Kerberos
+ * principal is accessed as that principal, and its token must not be handed
to the caller's job.
+ *
* @param renewer the renewer.
* @param credentials the credentials.
* @return the array of tokens.
*/
protected Token<?>[] addDelegationTokensForAllFS(String renewer, Credentials
credentials) {
+ String currentUser;
+ try {
+ currentUser = UserGroupInformation.getCurrentUser().getShortUserName();
+ } catch (IOException e) {
+ LOG.warn("Failed to get the current user, skipping delegation tokens",
e);
+ return new Token<?>[0];
+ }
+
List<Token<?>> tokenList = Lists.newArrayList();
for (FileSystem fileSystem : fileSystemCache.asMap().values()) {
Review Comment:
[Important] Token collection still depends on the filesystem cache retaining
every configured authority.
`prepareDelegationTokens()` resolves the whole list before this loop runs.
With `fs.gravitino.fileset.cache.maxCapacity=1` and two configured filesets on
different HDFS authorities, one entry can be evicted before collection, so only
one authority gets a token. The default capacity is 20, so this also affects
larger lists; expiration during preparation is another way to lose an entry.
Please collect tokens as each fileset is resolved, or collect from an
explicit set of the resolved filesystem instances rather than the bounded
cache. Retain collection for previously cached filesystems where needed for
compatibility.
Verified on 20b1ec7: a temporary regression test used the real GVFS cache
with capacity 1, two configured filesets, mocked filesystem construction/token
issuance, and cache cleanup between resolutions. It expected two credential
tokens and received one. This was not a real multi-cluster Kerberos E2E run.
Please add this boundary to the regression suite.
##########
clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/BaseGVFSOperations.java:
##########
@@ -453,23 +455,72 @@ public abstract FSDataOutputStream append(Path gvfsPath,
int bufferSize, Progres
*/
public abstract Token<?>[] addDelegationTokens(String renewer, Credentials
credentials);
+ /**
+ * Resolve the filesets listed in {@link
+ *
GravitinoVirtualFileSystemConfiguration#FS_GRAVITINO_DELEGATION_TOKEN_FILESETS},
so that their
+ * actual file systems exist before delegation tokens are collected. Token
requests carry no path,
+ * so the filesets cannot be learned from the caller.
+ */
+ public void prepareDelegationTokens() {
+ for (String path :
+ conf.getTrimmedStrings(
+
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_DELEGATION_TOKEN_FILESETS))
{
+ try {
+ getActualFileSystem(new Path(path), currentLocationName());
+ } catch (Exception e) {
+ // Never fail token collection: an unresolvable fileset simply yields
no token.
+ LOG.warn("Failed to resolve {} while preparing delegation tokens",
path, e);
+ }
+ }
+ }
+
/**
* Add delegation tokens for all file systems in the cache.
*
+ * <p>Only tokens owned by the current user are added. A fileset configured
with its own Kerberos
+ * principal is accessed as that principal, and its token must not be handed
to the caller's job.
+ *
* @param renewer the renewer.
* @param credentials the credentials.
* @return the array of tokens.
*/
protected Token<?>[] addDelegationTokensForAllFS(String renewer, Credentials
credentials) {
+ String currentUser;
+ try {
+ currentUser = UserGroupInformation.getCurrentUser().getShortUserName();
+ } catch (IOException e) {
+ LOG.warn("Failed to get the current user, skipping delegation tokens",
e);
+ return new Token<?>[0];
+ }
+
List<Token<?>> tokenList = Lists.newArrayList();
for (FileSystem fileSystem : fileSystemCache.asMap().values()) {
+ // Collect into a copy, so a token owned by another user never reaches
the caller.
+ Credentials fetched = new Credentials(credentials);
Review Comment:
[Important] This breaks the Hadoop `addDelegationTokens(renewer, null)`
contract.
Hadoop's `DelegationTokenIssuer.addDelegationTokens` accepts null
credentials, creates a local `Credentials`, and returns the newly acquired
tokens. GVFS's operations Javadoc explicitly refers to that contract. With a
populated cache, `new Credentials(null)` now throws a NullPointerException
before token collection. The later `credentials.getToken`/`addToken` calls
would also need the same normalization.
Please normalize null to a local empty `Credentials` at the method entry,
use it consistently for copying, deduplication and merging, and preserve the
returned token array.
Verified on 20b1ec7 with a temporary regression test whose mocked underlying
filesystem supports null as Hadoop does: the call failed in
`Credentials.addAll` with `other` null. Spark passes non-null credentials on
the reviewed path, so this is a public API compatibility regression rather than
the main Spark failure.
Hadoop implementation:
https://github.com/apache/hadoop/blob/rel/release-3.3.6/hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/token/DelegationTokenIssuer.java#L71
##########
clients/filesystem-hadoop3/src/main/java/org/apache/gravitino/filesystem/hadoop/BaseGVFSOperations.java:
##########
@@ -453,23 +455,72 @@ public abstract FSDataOutputStream append(Path gvfsPath,
int bufferSize, Progres
*/
public abstract Token<?>[] addDelegationTokens(String renewer, Credentials
credentials);
+ /**
+ * Resolve the filesets listed in {@link
+ *
GravitinoVirtualFileSystemConfiguration#FS_GRAVITINO_DELEGATION_TOKEN_FILESETS},
so that their
+ * actual file systems exist before delegation tokens are collected. Token
requests carry no path,
+ * so the filesets cannot be learned from the caller.
+ */
+ public void prepareDelegationTokens() {
+ for (String path :
+ conf.getTrimmedStrings(
+
GravitinoVirtualFileSystemConfiguration.FS_GRAVITINO_DELEGATION_TOKEN_FILESETS))
{
+ try {
+ getActualFileSystem(new Path(path), currentLocationName());
Review Comment:
[Important] Preparing tokens can now create the fileset's storage directory
before the job accesses any data.
This call reaches `getActualFileSystemByLocationName`, which calls
`createFilesetLocationIfNeed`. When the catalog has
`disable-filesystem-ops=true` and GVFS has
`fs.gravitino.autoCreateLocation=true` (the default), preparation performs
`exists()` and potentially `mkdirs()`. Token collection during submission or
renewal therefore introduces storage writes, even for a listed fileset that the
job never reads.
Please provide a resolution/initialization path for token preparation that
skips directory creation, leaving auto-creation on the data-operation path. Add
a regression asserting that preparing tokens does not create a missing location.
Verified on 20b1ec7 using the real local filesystem and the production
resolution path, with mocked Gravitino metadata: calling only
`prepareDelegationTokens()` created an initially absent fileset directory. HDFS
permission/error behavior was not executed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
