This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 3a75387e613 [fix](catalog) re-auth Iceberg REST catalog on 401 instead
of wedging until FE restart (#64966)
3a75387e613 is described below
commit 3a75387e61388bd886eeef38b794d5ed4eb298bf
Author: Raghvendra Singh <[email protected]>
AuthorDate: Fri Jul 17 11:23:52 2026 +0530
[fix](catalog) re-auth Iceberg REST catalog on 401 instead of wedging until
FE restart (#64966)
### What problem does this PR solve?
Problem Summary:
An Iceberg REST catalog using oauth2 client credentials obtains its
token when the catalog client is built and keeps it refreshed in the
background. If a refresh permanently fails (for example the auth server
is briefly unreachable or overloaded right when the refresh fires), the
client is left holding a stale token, and every subsequent request fails
with `NotAuthorizedException: Not authorized` (HTTP 401) — forever. The
client has no re-authentication path of its own, and neither `REFRESH
CATALOG` nor metadata-cache invalidation rebuilds it, so the catalog
stays unusable until the FE process is restarted.
We hit this in production against an Apache Polaris REST catalog. The
failure mode is nasty to diagnose because the 401 is swallowed:
`getDbNullable`/`buildDbForInit` report `Database [<db>] does not
exist`, `SHOW DATABASES` returns a partial list, and since DDL
statements (`CREATE VIEW`/`CREATE MTMV`/CTAS/`INSERT`) are
`ForwardToMaster`, one wedged master FE breaks all DDL cluster-wide
while plain SELECTs on follower FEs keep working. Only an FE restart
recovered.
### What is changed?
Recovery now lives entirely at the REST layer (per review feedback —
thanks @CalvinKirs): a new `ReauthenticatingRestSessionCatalog` (a
`BaseViewSessionCatalog`) that `IcebergRestProperties` wraps around the
`RESTSessionCatalog` it already builds and owns. On a 401 raised by a
request made under the catalog's own identity, the wrapper rebuilds the
delegate through the same build seam (fresh HTTP client, fresh OAuth2
token fetch), closes the wedged client, and retries the operation once.
If the retry also fails, the error propagates unchanged.
Design notes:
- **REST-only.** Only `IcebergRestProperties` constructs the wrapper; no
other catalog type sees this logic. `IcebergMetadataOps` is untouched
apart from widening two declared types from `RESTSessionCatalog` to
`BaseViewSessionCatalog`.
- **All paths covered, no call-site changes, no stale references.** The
default catalog (`asCatalog(empty)`), the view catalog
(`asViewCatalog(empty)`), and per-user delegated sessions are all thin
views that call back into the session catalog, so they inherit the
recovery automatically — and because the swap happens below them, no
holder of any of those references ever points at a dead client after
recovery.
- **Reads and mutations both retried.** A 401 is rejected by the server
before the request is processed, so retrying after re-authentication
cannot double-apply an operation.
- **Per-user delegated sessions are excluded.** A 401 on a request
carrying a delegated (per-user) credential means that user's token is
invalid; rebuilding the shared client cannot fix it and must not be
triggered by it. Likewise, when catalog initialization itself runs under
a delegated credential (`iceberg.rest.session=user`), no wrapper is
installed, so the rebuild path can never capture a user token into the
shared client.
- **Typed detection only.** `NotAuthorizedException` anywhere in the
cause chain; no message-string matching.
- **Bounded.** One rebuild + one retry per failed request; concurrent
401s coalesce into a single rebuild (a request that raced with a
completed rebuild just retries on the fresh client).
Boundary: recovery triggers on catalog-level operations. Objects already
handed out (a loaded `Table`/`View`, a `buildTable`/`buildView` builder)
keep the client they were created with; if the token expires between
obtaining and using one, that single use can still 401, and the next
catalog-level operation heals the client.
### Behavior changed:
- Iceberg REST catalogs recover automatically from an expired/rejected
OAuth2 token (401) by re-authenticating and retrying once, instead of
failing every request until the FE restarts. Other catalog types and
non-401 failures are unaffected.
Signed-off-by: Raghvendra Singh <[email protected]>
Co-authored-by: Claude <[email protected]>
---
.../datasource/iceberg/IcebergMetadataOps.java | 6 +-
.../iceberg/IcebergRestExternalCatalog.java | 4 +-
.../iceberg/IcebergUserSessionCatalog.java | 4 +-
.../ReauthenticatingRestSessionCatalog.java | 283 +++++++++++++++++++++
.../property/metastore/IcebergRestProperties.java | 33 ++-
.../ReauthenticatingRestSessionCatalogTest.java | 191 ++++++++++++++
6 files changed, 508 insertions(+), 13 deletions(-)
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 b3cfc172731..c8d9c50c21d 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
@@ -69,6 +69,7 @@ import org.apache.iceberg.Table;
import org.apache.iceberg.TableProperties;
import org.apache.iceberg.UpdatePartitionSpec;
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;
@@ -79,7 +80,6 @@ import org.apache.iceberg.exceptions.NoSuchNamespaceException;
import org.apache.iceberg.expressions.Expressions;
import org.apache.iceberg.expressions.Literal;
import org.apache.iceberg.expressions.Term;
-import org.apache.iceberg.rest.RESTSessionCatalog;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import org.apache.iceberg.types.Types.NestedField;
@@ -134,7 +134,7 @@ public class IcebergMetadataOps implements
ExternalMetadataOps {
// properties. For any other catalog type this is null (session
disabled).
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
@@ -1383,7 +1383,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 c3d7a88bf01..44f194a0bf7 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;
@@ -226,7 +226,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 d420e40724f..d276b6c317d 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.S3Properties;
@@ -32,6 +33,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;
@@ -41,6 +43,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;
@@ -66,7 +69,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"},
@@ -241,7 +247,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());
}
@@ -263,7 +284,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;
}
@@ -272,14 +293,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]
