This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 1a4a1e60ed9 branch-4.2: [fix](catalog) re-auth Iceberg REST catalog on
401 instead of wedging until FE restart (#70030)
1a4a1e60ed9 is described below
commit 1a4a1e60ed9622340b3308b4ebae57ab4a097c0a
Author: Calvin Kirs <[email protected]>
AuthorDate: Thu Nov 5 09:42:29 2026 +0800
branch-4.2: [fix](catalog) re-auth Iceberg REST catalog on 401 instead of
wedging until FE restart (#70030)
https://github.com/apache/doris/pull/64966
Co-authored-by: Raghvendra Singh <[email protected]>
---
.../datasource/iceberg/IcebergExternalCatalog.java | 4 +-
.../datasource/iceberg/IcebergMetadataOps.java | 6 +-
.../iceberg/IcebergRestExternalCatalog.java | 4 +-
.../iceberg/IcebergUserSessionCatalog.java | 4 +-
.../ReauthenticatingRestSessionCatalog.java | 283 +++++++++++++++++++++
.../property/metastore/IcebergRestProperties.java | 33 ++-
.../ReauthenticatingRestSessionCatalogTest.java | 191 ++++++++++++++
7 files changed, 510 insertions(+), 15 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalog.java
index 71a8de40dc7..9e3001386a1 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalog.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergExternalCatalog.java
@@ -43,8 +43,8 @@ import
org.apache.doris.foundation.security.JdbcDriverUrlSecurity;
import org.apache.doris.transaction.TransactionManagerFactory;
import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.BaseViewSessionCatalog;
import org.apache.iceberg.catalog.Catalog;
-import org.apache.iceberg.rest.RESTSessionCatalog;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -286,7 +286,7 @@ public abstract class IcebergExternalCatalog extends
ExternalCatalog {
ExternalMetaCacheMgr cacheMgr =
Env.getCurrentEnv().getExtMetaCacheMgr();
cacheMgr.removeCatalogByEngine(getId(),
IcebergExternalMetaCache.ENGINE);
Catalog retiredCatalog = catalog;
- RESTSessionCatalog retiredRestCatalog = msProperties instanceof
IcebergRestProperties
+ BaseViewSessionCatalog retiredRestCatalog = msProperties instanceof
IcebergRestProperties
? ((IcebergRestProperties)
msProperties).getRestSessionCatalog() : null;
catalog = null;
resourceTracker.retireCurrent(() -> {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java
index 728f4a1c3cd..b200b40ea46 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergMetadataOps.java
@@ -72,6 +72,7 @@ import org.apache.iceberg.TableProperties;
import org.apache.iceberg.UpdatePartitionSpec;
import org.apache.iceberg.UpdateProperties;
import org.apache.iceberg.UpdateSchema;
+import org.apache.iceberg.catalog.BaseViewSessionCatalog;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.Namespace;
import org.apache.iceberg.catalog.SessionCatalog;
@@ -86,7 +87,6 @@ import org.apache.iceberg.expressions.Term;
import org.apache.iceberg.io.FileIO;
import org.apache.iceberg.io.LocationProvider;
import org.apache.iceberg.jdbc.JdbcCatalog;
-import org.apache.iceberg.rest.RESTSessionCatalog;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.TypeUtil;
import org.apache.iceberg.types.Types;
@@ -151,7 +151,7 @@ public class IcebergMetadataOps implements
ExternalMetadataOps {
// Other catalog types have no session-aware capability.
this.userSessionCatalog = dorisCatalog instanceof
IcebergUserSessionCatalog
? (IcebergUserSessionCatalog) dorisCatalog : null;
- RESTSessionCatalog restSessionCatalog =
+ BaseViewSessionCatalog restSessionCatalog =
userSessionCatalog == null ? null :
userSessionCatalog.getRestSessionCatalog();
IcebergRestProperties.DelegatedTokenMode delegatedTokenMode =
userSessionCatalog == null ?
IcebergRestProperties.DelegatedTokenMode.ACCESS_TOKEN
@@ -2193,7 +2193,7 @@ public class IcebergMetadataOps implements
ExternalMetadataOps {
return defaultViewCatalog;
}
- private Optional<ViewCatalog> resolveDefaultViewCatalog(Catalog catalog,
RESTSessionCatalog restSessionCatalog,
+ private Optional<ViewCatalog> resolveDefaultViewCatalog(Catalog catalog,
BaseViewSessionCatalog restSessionCatalog,
boolean viewEnabled) {
// Branch on whether this is a REST (session-aware) catalog, not on
whether restSessionCatalog happens to
// be built: for REST the default Catalog (asCatalog) is not a
ViewCatalog, so views must come from the
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergRestExternalCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergRestExternalCatalog.java
index 3a70e945180..431941365eb 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergRestExternalCatalog.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergRestExternalCatalog.java
@@ -29,7 +29,7 @@ import
org.apache.doris.datasource.property.metastore.IcebergRestProperties;
import org.apache.doris.datasource.property.metastore.MetastoreProperties;
import com.google.common.collect.Lists;
-import org.apache.iceberg.rest.RESTSessionCatalog;
+import org.apache.iceberg.catalog.BaseViewSessionCatalog;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
@@ -213,7 +213,7 @@ public class IcebergRestExternalCatalog extends
IcebergExternalCatalog implement
}
@Override
- public RESTSessionCatalog getRestSessionCatalog() {
+ public BaseViewSessionCatalog getRestSessionCatalog() {
IcebergRestProperties props = restProperties();
return props == null ? null : props.getRestSessionCatalog();
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUserSessionCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUserSessionCatalog.java
index 4aae96ead4a..e0d8b1c5b39 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUserSessionCatalog.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/IcebergUserSessionCatalog.java
@@ -20,7 +20,7 @@ package org.apache.doris.datasource.iceberg;
import org.apache.doris.datasource.SessionContext;
import org.apache.doris.datasource.property.metastore.IcebergRestProperties;
-import org.apache.iceberg.rest.RESTSessionCatalog;
+import org.apache.iceberg.catalog.BaseViewSessionCatalog;
/**
* Capability interface for an Iceberg catalog that supports per-user dynamic
session
@@ -47,7 +47,7 @@ public interface IcebergUserSessionCatalog {
boolean useSessionCatalog(SessionContext ctx);
/** The session-aware Iceberg REST catalog backing this catalog (may be
null before initialization). */
- RESTSessionCatalog getRestSessionCatalog();
+ BaseViewSessionCatalog getRestSessionCatalog();
/** The delegated-token mode used when attaching the user's credential to
session requests. */
IcebergRestProperties.DelegatedTokenMode getDelegatedTokenMode();
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/ReauthenticatingRestSessionCatalog.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/ReauthenticatingRestSessionCatalog.java
new file mode 100644
index 00000000000..c39b0a73381
--- /dev/null
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/iceberg/ReauthenticatingRestSessionCatalog.java
@@ -0,0 +1,283 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.iceberg;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.apache.commons.lang3.exception.ExceptionUtils;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.Table;
+import org.apache.iceberg.catalog.BaseViewSessionCatalog;
+import org.apache.iceberg.catalog.Catalog;
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SessionCatalog.SessionContext;
+import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.exceptions.NotAuthorizedException;
+import org.apache.iceberg.rest.RESTSessionCatalog;
+import org.apache.iceberg.view.View;
+import org.apache.iceberg.view.ViewBuilder;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Supplier;
+
+/**
+ * A session catalog that transparently recovers from an expired/rejected
credential on an Iceberg REST
+ * catalog, instead of failing every request until the FE restarts.
+ *
+ * <p><b>Why this exists.</b> {@link RESTSessionCatalog} obtains an OAuth2
token at {@code initialize()} and
+ * keeps it refreshed in the background. If a refresh permanently fails (e.g.
the auth server was briefly
+ * unreachable or overloaded at refresh time), the client is left holding a
stale token and every subsequent
+ * request fails with {@link NotAuthorizedException} (HTTP 401) forever — the
client has no re-authentication
+ * path of its own, and neither {@code REFRESH CATALOG} nor metadata-cache
invalidation rebuilds it. The only
+ * recovery is building a fresh client, which is exactly what this wrapper
does: on a 401 from a request made
+ * under the catalog's own identity, it rebuilds the delegate {@link
RESTSessionCatalog} (forcing a fresh
+ * OAuth2 token fetch), closes the wedged one, and retries the operation once.
If the retry also fails, the
+ * error propagates unchanged.
+ *
+ * <p><b>Why a wrapper at the session-catalog level.</b> Everything Doris
hands out for a REST catalog — the
+ * default {@code asCatalog(empty)} used by {@code IcebergMetadataOps}, the
{@code asViewCatalog(empty)} view
+ * path, and per-user delegated sessions — is a thin view that calls back into
this class (see
+ * {@link org.apache.iceberg.catalog.BaseSessionCatalog.AsCatalog}). Wrapping
here means no holder of any of
+ * those references ever sees a stale client after recovery, and non-REST
catalogs are untouched because only
+ * {@code IcebergRestProperties} constructs this class.
+ *
+ * <p><b>What is retried.</b> Both reads and mutations: a 401 is rejected by
the server before the request is
+ * processed, so retrying after re-authentication cannot double-apply an
operation. Requests that carry a
+ * per-user delegated credential are <b>not</b> recovered — a 401 there means
that user's token is invalid,
+ * and rebuilding the shared client cannot (and must not) fix it.
+ *
+ * <p><b>Boundary.</b> Recovery triggers on catalog-level operations. Objects
already handed out (a loaded
+ * {@link Table}/{@link View}, a builder from {@code buildTable}/{@code
buildView}) keep their own reference
+ * to the client they were created with; if the credential expires between
obtaining such an object and using
+ * it, that one use can still fail with a 401. The next catalog-level
operation rebuilds the client, after
+ * which reloading the object succeeds.
+ */
+public class ReauthenticatingRestSessionCatalog extends BaseViewSessionCatalog
implements Closeable {
+
+ private static final Logger LOG =
LogManager.getLogger(ReauthenticatingRestSessionCatalog.class);
+
+ private final Supplier<RESTSessionCatalog> delegateBuilder;
+ private volatile RESTSessionCatalog delegate;
+
+ public ReauthenticatingRestSessionCatalog(RESTSessionCatalog
initialDelegate,
+ Supplier<RESTSessionCatalog> delegateBuilder) {
+ this.delegate = initialDelegate;
+ this.delegateBuilder = delegateBuilder;
+ }
+
+ @VisibleForTesting
+ RESTSessionCatalog currentDelegate() {
+ return delegate;
+ }
+
+ private <T> T withAuthRecovery(SessionContext context, Supplier<T> op) {
+ RESTSessionCatalog attemptedOn = delegate;
+ try {
+ return op.get();
+ } catch (RuntimeException e) {
+ if (!isAuthExpired(e) || !usesCatalogIdentity(context)) {
+ throw e;
+ }
+ reauthenticate(attemptedOn, e);
+ return op.get();
+ }
+ }
+
+ private void runWithAuthRecovery(SessionContext context, Runnable op) {
+ withAuthRecovery(context, () -> {
+ op.run();
+ return null;
+ });
+ }
+
+ /**
+ * Rebuilds the delegate (fresh client, fresh token) and closes the wedged
one. Synchronized so that
+ * concurrent 401s coalesce into a single rebuild: a thread whose failed
attempt ran against an
+ * already-replaced delegate skips the rebuild and just retries on the
fresh one.
+ */
+ private synchronized void reauthenticate(RESTSessionCatalog attemptedOn,
RuntimeException cause) {
+ if (delegate != attemptedOn) {
+ return;
+ }
+ LOG.warn("Iceberg REST catalog {} rejected its cached credential (401
Not Authorized) and the client "
+ + "cannot recover it internally. Rebuilding the REST client to
force re-authentication, "
+ + "then retrying the request once.", name(), cause);
+ RESTSessionCatalog replacement = delegateBuilder.get();
+ RESTSessionCatalog wedged = delegate;
+ delegate = replacement;
+ try {
+ wedged.close();
+ } catch (IOException | RuntimeException e) {
+ LOG.warn("Failed to close the replaced Iceberg REST client of
catalog {}", name(), e);
+ }
+ }
+
+ /**
+ * A request made with a per-user delegated credential authenticates as
that user, not as the catalog;
+ * a 401 there is that token's problem and must not rebuild (or leak into)
the shared client.
+ */
+ private static boolean usesCatalogIdentity(SessionContext context) {
+ return context == null || context.credentials() == null ||
context.credentials().isEmpty();
+ }
+
+ private static boolean isAuthExpired(Throwable t) {
+ return ExceptionUtils.getThrowableList(t).stream().anyMatch(c -> c
instanceof NotAuthorizedException);
+ }
+
+ // ---- SessionCatalog ----
+
+ @Override
+ public void initialize(String name, Map<String, String> properties) {
+ super.initialize(name, properties);
+ delegate.initialize(name, properties);
+ }
+
+ @Override
+ public String name() {
+ return delegate.name();
+ }
+
+ @Override
+ public Map<String, String> properties() {
+ return delegate.properties();
+ }
+
+ @Override
+ public List<TableIdentifier> listTables(SessionContext context, Namespace
ns) {
+ return withAuthRecovery(context, () -> delegate.listTables(context,
ns));
+ }
+
+ @Override
+ public Catalog.TableBuilder buildTable(SessionContext context,
TableIdentifier ident, Schema schema) {
+ return withAuthRecovery(context, () -> delegate.buildTable(context,
ident, schema));
+ }
+
+ @Override
+ public Table registerTable(SessionContext context, TableIdentifier ident,
String metadataFileLocation) {
+ return withAuthRecovery(context, () -> delegate.registerTable(context,
ident, metadataFileLocation));
+ }
+
+ @Override
+ public boolean tableExists(SessionContext context, TableIdentifier ident) {
+ return withAuthRecovery(context, () -> delegate.tableExists(context,
ident));
+ }
+
+ @Override
+ public Table loadTable(SessionContext context, TableIdentifier ident) {
+ return withAuthRecovery(context, () -> delegate.loadTable(context,
ident));
+ }
+
+ @Override
+ public boolean dropTable(SessionContext context, TableIdentifier ident) {
+ return withAuthRecovery(context, () -> delegate.dropTable(context,
ident));
+ }
+
+ @Override
+ public boolean purgeTable(SessionContext context, TableIdentifier ident) {
+ return withAuthRecovery(context, () -> delegate.purgeTable(context,
ident));
+ }
+
+ @Override
+ public void renameTable(SessionContext context, TableIdentifier from,
TableIdentifier to) {
+ runWithAuthRecovery(context, () -> delegate.renameTable(context, from,
to));
+ }
+
+ @Override
+ public void invalidateTable(SessionContext context, TableIdentifier ident)
{
+ runWithAuthRecovery(context, () -> delegate.invalidateTable(context,
ident));
+ }
+
+ @Override
+ public void createNamespace(SessionContext context, Namespace namespace,
Map<String, String> metadata) {
+ runWithAuthRecovery(context, () -> delegate.createNamespace(context,
namespace, metadata));
+ }
+
+ @Override
+ public List<Namespace> listNamespaces(SessionContext context, Namespace
namespace) {
+ return withAuthRecovery(context, () ->
delegate.listNamespaces(context, namespace));
+ }
+
+ @Override
+ public Map<String, String> loadNamespaceMetadata(SessionContext context,
Namespace namespace) {
+ return withAuthRecovery(context, () ->
delegate.loadNamespaceMetadata(context, namespace));
+ }
+
+ @Override
+ public boolean dropNamespace(SessionContext context, Namespace namespace) {
+ return withAuthRecovery(context, () -> delegate.dropNamespace(context,
namespace));
+ }
+
+ @Override
+ public boolean updateNamespaceMetadata(SessionContext context, Namespace
namespace,
+ Map<String, String> updates, Set<String> removals) {
+ return withAuthRecovery(context, () ->
delegate.updateNamespaceMetadata(context, namespace, updates,
+ removals));
+ }
+
+ @Override
+ public boolean namespaceExists(SessionContext context, Namespace
namespace) {
+ return withAuthRecovery(context, () ->
delegate.namespaceExists(context, namespace));
+ }
+
+ // ---- ViewSessionCatalog ----
+
+ @Override
+ public List<TableIdentifier> listViews(SessionContext context, Namespace
namespace) {
+ return withAuthRecovery(context, () -> delegate.listViews(context,
namespace));
+ }
+
+ @Override
+ public View loadView(SessionContext context, TableIdentifier identifier) {
+ return withAuthRecovery(context, () -> delegate.loadView(context,
identifier));
+ }
+
+ @Override
+ public boolean viewExists(SessionContext context, TableIdentifier
identifier) {
+ return withAuthRecovery(context, () -> delegate.viewExists(context,
identifier));
+ }
+
+ @Override
+ public ViewBuilder buildView(SessionContext context, TableIdentifier
identifier) {
+ return withAuthRecovery(context, () -> delegate.buildView(context,
identifier));
+ }
+
+ @Override
+ public boolean dropView(SessionContext context, TableIdentifier
identifier) {
+ return withAuthRecovery(context, () -> delegate.dropView(context,
identifier));
+ }
+
+ @Override
+ public void renameView(SessionContext context, TableIdentifier from,
TableIdentifier to) {
+ runWithAuthRecovery(context, () -> delegate.renameView(context, from,
to));
+ }
+
+ @Override
+ public void invalidateView(SessionContext context, TableIdentifier
identifier) {
+ runWithAuthRecovery(context, () -> delegate.invalidateView(context,
identifier));
+ }
+
+ @Override
+ public void close() throws IOException {
+ delegate.close();
+ }
+}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/IcebergRestProperties.java
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/IcebergRestProperties.java
index eee0fc52972..01ac81eeb2d 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/IcebergRestProperties.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/datasource/property/metastore/IcebergRestProperties.java
@@ -21,6 +21,7 @@ import org.apache.doris.datasource.DelegatedCredential;
import org.apache.doris.datasource.SessionContext;
import org.apache.doris.datasource.iceberg.IcebergDelegatedCredentialUtils;
import org.apache.doris.datasource.iceberg.IcebergExternalCatalog;
+import org.apache.doris.datasource.iceberg.ReauthenticatingRestSessionCatalog;
import org.apache.doris.datasource.property.common.AwsCredentialsProviderMode;
import
org.apache.doris.datasource.property.common.IcebergAwsClientCredentialsProperties;
import org.apache.doris.datasource.property.storage.OSSProperties;
@@ -33,6 +34,7 @@ import lombok.Getter;
import org.apache.hadoop.conf.Configuration;
import org.apache.iceberg.CatalogProperties;
import org.apache.iceberg.CatalogUtil;
+import org.apache.iceberg.catalog.BaseViewSessionCatalog;
import org.apache.iceberg.catalog.Catalog;
import org.apache.iceberg.catalog.SessionCatalog;
import org.apache.iceberg.rest.RESTSessionCatalog;
@@ -42,6 +44,7 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.util.Strings;
+import java.io.Closeable;
import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
@@ -68,7 +71,10 @@ public class IcebergRestProperties extends
AbstractIcebergProperties {
// asCatalog(SessionContext) / asViewCatalog(SessionContext), without
reflecting RESTCatalog's private
// sessionCatalog field. This is the single underlying catalog shared by
the default and user-session
// paths; IcebergMetadataOps reads it via getRestSessionCatalog() and owns
no other REST catalog.
- private RESTSessionCatalog restSessionCatalog;
+ // When the catalog authenticates with its own identity this is a
ReauthenticatingRestSessionCatalog
+ // wrapping the RESTSessionCatalog, so an expired/rejected token is
recovered by rebuilding the client
+ // instead of failing every request until the FE restarts.
+ private BaseViewSessionCatalog restSessionCatalog;
@Getter
@ConnectorProperty(names = {"iceberg.rest.uri", "uri"},
@@ -243,7 +249,22 @@ public class IcebergRestProperties extends
AbstractIcebergProperties {
// it ourselves keeps that capability without reflection. The "type"
key is dropped because the
// Iceberg SDK rejects "type" together with a concrete catalog impl.
catalogProps.remove(CatalogUtil.ICEBERG_CATALOG_TYPE);
- this.restSessionCatalog = buildRestSessionCatalog(catalogName,
catalogProps, configuration);
+ RESTSessionCatalog rawSessionCatalog =
buildRestSessionCatalog(catalogName, catalogProps, configuration);
+ if (sessionContext == null ||
!sessionContext.hasDelegatedCredential()) {
+ // The catalog authenticates with its own identity (e.g. an oauth2
client credential). If that
+ // credential's token expires and the client cannot refresh it
(auth server briefly unreachable at
+ // refresh time), the RESTSessionCatalog is left rejecting every
request with a 401 until the FE
+ // restarts. Wrap it so a 401 rebuilds the client (fresh token)
and retries once. The rebuild
+ // supplier re-resolves from the same catalog-identity properties,
so it can never capture a
+ // per-user delegated credential.
+ Map<String, String> frozenProps = Collections.unmodifiableMap(new
HashMap<>(catalogProps));
+ this.restSessionCatalog = new
ReauthenticatingRestSessionCatalog(rawSessionCatalog,
+ () -> buildRestSessionCatalog(catalogName, frozenProps,
configuration));
+ } else {
+ // Catalog initialization under a per-user delegated credential:
recovery must not re-mint the
+ // client with that user's token, so no wrapper — behavior is
unchanged from before.
+ this.restSessionCatalog = rawSessionCatalog;
+ }
// The default (non-delegated) Catalog is asCatalog(empty), identical
to what RESTCatalog exposes.
return
restSessionCatalog.asCatalog(SessionCatalog.SessionContext.createEmpty());
}
@@ -265,7 +286,7 @@ public class IcebergRestProperties extends
AbstractIcebergProperties {
* catalog has not been initialized yet. Callers use it to obtain
per-request catalogs via
* {@code asCatalog(SessionContext)} / {@code
asViewCatalog(SessionContext)}.
*/
- public RESTSessionCatalog getRestSessionCatalog() {
+ public BaseViewSessionCatalog getRestSessionCatalog() {
return restSessionCatalog;
}
@@ -274,14 +295,14 @@ public class IcebergRestProperties extends
AbstractIcebergProperties {
* Safe to call multiple times and before initialization.
*/
public void closeRestSessionCatalog() {
- if (restSessionCatalog != null) {
+ if (restSessionCatalog instanceof Closeable) {
try {
- restSessionCatalog.close();
+ ((Closeable) restSessionCatalog).close();
} catch (IOException e) {
LOG.warn("Failed to close Iceberg REST session catalog", e);
}
- restSessionCatalog = null;
}
+ restSessionCatalog = null;
}
@Override
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/ReauthenticatingRestSessionCatalogTest.java
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/ReauthenticatingRestSessionCatalogTest.java
new file mode 100644
index 00000000000..2132a38febb
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/datasource/iceberg/ReauthenticatingRestSessionCatalogTest.java
@@ -0,0 +1,191 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.datasource.iceberg;
+
+import org.apache.iceberg.catalog.Namespace;
+import org.apache.iceberg.catalog.SessionCatalog.SessionContext;
+import org.apache.iceberg.exceptions.NotAuthorizedException;
+import org.apache.iceberg.rest.RESTSessionCatalog;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class ReauthenticatingRestSessionCatalogTest {
+
+ private static final Namespace NS = Namespace.empty();
+
+ /**
+ * A stand-in for a live REST catalog: throws the configured failure on
every namespace listing until the
+ * failure is cleared, and records whether it was closed. All overridden
methods avoid the real
+ * RESTSessionCatalog internals, so no initialization or network is
involved.
+ */
+ private static final class FakeRestSessionCatalog extends
RESTSessionCatalog {
+ private final String label;
+ private final RuntimeException failure;
+ private final AtomicInteger listCalls = new AtomicInteger();
+ private volatile boolean closed;
+
+ FakeRestSessionCatalog(String label, RuntimeException failure) {
+ this.label = label;
+ this.failure = failure;
+ }
+
+ @Override
+ public String name() {
+ return label;
+ }
+
+ @Override
+ public List<Namespace> listNamespaces(SessionContext context,
Namespace ns) {
+ listCalls.incrementAndGet();
+ if (failure != null) {
+ throw failure;
+ }
+ return Collections.singletonList(Namespace.of(label));
+ }
+
+ @Override
+ public void close() {
+ closed = true;
+ }
+ }
+
+ private static NotAuthorizedException notAuthorized() {
+ return new NotAuthorizedException("Not authorized: %s", "the token
expired");
+ }
+
+ private static SessionContext delegatedUserContext() {
+ return new SessionContext(UUID.randomUUID().toString(), "alice",
+ Collections.singletonMap("token", "user-token"),
Collections.emptyMap());
+ }
+
+ @Test
+ public void testNotAuthorizedRebuildsClientAndRetriesOnce() {
+ FakeRestSessionCatalog wedged = new FakeRestSessionCatalog("wedged",
notAuthorized());
+ FakeRestSessionCatalog fresh = new FakeRestSessionCatalog("fresh",
null);
+ AtomicInteger rebuilds = new AtomicInteger();
+ ReauthenticatingRestSessionCatalog catalog = new
ReauthenticatingRestSessionCatalog(wedged, () -> {
+ rebuilds.incrementAndGet();
+ return fresh;
+ });
+
+ List<Namespace> namespaces =
catalog.listNamespaces(SessionContext.createEmpty(), NS);
+
+
Assertions.assertEquals(Collections.singletonList(Namespace.of("fresh")),
namespaces);
+ Assertions.assertEquals(1, rebuilds.get());
+ Assertions.assertEquals(1, wedged.listCalls.get());
+ Assertions.assertEquals(1, fresh.listCalls.get());
+ Assertions.assertTrue(wedged.closed, "the wedged client must be closed
after replacement");
+ Assertions.assertSame(fresh, catalog.currentDelegate());
+ }
+
+ @Test
+ public void testNotAuthorizedWrappedInAnotherExceptionIsDetected() {
+ FakeRestSessionCatalog wedged = new FakeRestSessionCatalog("wedged",
+ new RuntimeException("Failed to list database names",
notAuthorized()));
+ FakeRestSessionCatalog fresh = new FakeRestSessionCatalog("fresh",
null);
+ ReauthenticatingRestSessionCatalog catalog =
+ new ReauthenticatingRestSessionCatalog(wedged, () -> fresh);
+
+ List<Namespace> namespaces =
catalog.listNamespaces(SessionContext.createEmpty(), NS);
+
+
Assertions.assertEquals(Collections.singletonList(Namespace.of("fresh")),
namespaces);
+ }
+
+ @Test
+ public void testStillNotAuthorizedAfterRebuildPropagatesWithoutLooping() {
+ FakeRestSessionCatalog wedged = new FakeRestSessionCatalog("wedged",
notAuthorized());
+ FakeRestSessionCatalog stillWedged = new
FakeRestSessionCatalog("still-wedged", notAuthorized());
+ AtomicInteger rebuilds = new AtomicInteger();
+ ReauthenticatingRestSessionCatalog catalog = new
ReauthenticatingRestSessionCatalog(wedged, () -> {
+ rebuilds.incrementAndGet();
+ return stillWedged;
+ });
+
+ Assertions.assertThrows(NotAuthorizedException.class,
+ () -> catalog.listNamespaces(SessionContext.createEmpty(),
NS));
+ Assertions.assertEquals(1, rebuilds.get(), "exactly one rebuild, no
retry loop");
+ Assertions.assertEquals(1, wedged.listCalls.get());
+ Assertions.assertEquals(1, stillWedged.listCalls.get());
+ }
+
+ @Test
+ public void testNonAuthFailuresAreNotRetried() {
+ FakeRestSessionCatalog failing = new FakeRestSessionCatalog("failing",
+ new RuntimeException("connection reset"));
+ AtomicInteger rebuilds = new AtomicInteger();
+ ReauthenticatingRestSessionCatalog catalog = new
ReauthenticatingRestSessionCatalog(failing, () -> {
+ rebuilds.incrementAndGet();
+ return new FakeRestSessionCatalog("fresh", null);
+ });
+
+ Assertions.assertThrows(RuntimeException.class,
+ () -> catalog.listNamespaces(SessionContext.createEmpty(),
NS));
+ Assertions.assertEquals(0, rebuilds.get());
+ Assertions.assertEquals(1, failing.listCalls.get());
+ Assertions.assertFalse(failing.closed);
+ }
+
+ @Test
+ public void testDelegatedUserSessionIsNotRecovered() {
+ // A 401 for a request carrying a per-user delegated credential means
that user's token is invalid.
+ // Rebuilding the shared client cannot fix it and must not be
triggered by it.
+ FakeRestSessionCatalog wedged = new FakeRestSessionCatalog("wedged",
notAuthorized());
+ AtomicInteger rebuilds = new AtomicInteger();
+ ReauthenticatingRestSessionCatalog catalog = new
ReauthenticatingRestSessionCatalog(wedged, () -> {
+ rebuilds.incrementAndGet();
+ return new FakeRestSessionCatalog("fresh", null);
+ });
+
+ Assertions.assertThrows(NotAuthorizedException.class,
+ () -> catalog.listNamespaces(delegatedUserContext(), NS));
+ Assertions.assertEquals(0, rebuilds.get());
+ Assertions.assertFalse(wedged.closed);
+ }
+
+ @Test
+ public void testAsCatalogViewRoutesThroughRecovery() {
+ // The default Catalog handed to IcebergMetadataOps is
asCatalog(empty); it must inherit the same
+ // recovery because it calls back into this session catalog.
+ FakeRestSessionCatalog wedged = new FakeRestSessionCatalog("wedged",
notAuthorized());
+ FakeRestSessionCatalog fresh = new FakeRestSessionCatalog("fresh",
null);
+ ReauthenticatingRestSessionCatalog catalog =
+ new ReauthenticatingRestSessionCatalog(wedged, () -> fresh);
+
+ List<Namespace> namespaces =
((org.apache.iceberg.catalog.SupportsNamespaces)
+
catalog.asCatalog(SessionContext.createEmpty())).listNamespaces(NS);
+
+
Assertions.assertEquals(Collections.singletonList(Namespace.of("fresh")),
namespaces);
+ Assertions.assertTrue(wedged.closed);
+ }
+
+ @Test
+ public void testCloseClosesCurrentDelegate() throws Exception {
+ FakeRestSessionCatalog delegate = new
FakeRestSessionCatalog("delegate", null);
+ ReauthenticatingRestSessionCatalog catalog =
+ new ReauthenticatingRestSessionCatalog(delegate, () ->
delegate);
+
+ catalog.close();
+
+ Assertions.assertTrue(delegate.closed);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
