github-actions[bot] commented on code in PR #64966: URL: https://github.com/apache/doris/pull/64966#discussion_r3600562778
########## 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; Review Comment: [P2] Bind the operation to the delegate recorded for this attempt `attemptedOn` is read here, but the supplied lambda dereferences the volatile `delegate` again later. If another thread replaces D0 with D1 between those reads, this request actually runs on D1; when D1 returns 401, `reauthenticate(D0, ...)` sees `delegate != attemptedOn`, skips rebuilding D1, and line 102 simply retries the same rejected D1. Pass the selected delegate into the operation (for example, a `Function<RESTSessionCatalog, T>`) so the generation used for coalescing is exactly the generation that failed, and add a deterministic swap-between-snapshot-and-invocation test. ########## 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(); Review Comment: [P1] Prevent recovery from publishing after catalog close/reset `ExternalCatalog.resetToUninitialized()` closes this wrapper through `IcebergRestExternalCatalog.onClose()` under the external-catalog monitor, but `close()` does not synchronize with this recovery or mark the wrapper closed. If reset runs while `delegateBuilder.get()` is fetching config/token, it can close the old delegate and clear `IcebergRestProperties.restSessionCatalog`; this thread then publishes a fresh delegate into the discarded wrapper and retries with the frozen old properties. That replacement has no owner left to close, and a mutation can run after teardown/property change. Add a closed/generation invariant shared by `close` and recovery: a close that wins must prevent publication/retry, and any just-built losing replacement must be closed. Please cover the build-versus-reset interleavings with latch-based tests. ########## 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(); Review Comment: [P1] Retire the old delegate only after evicting or draining its users This closes more than the client that returned 401. Iceberg's loaded `Table`/`View` objects retain child HTTP clients and FileIO from this delegate, while Doris stores those same objects in `IcebergExternalMetaCache`; `IcebergTransaction.beginInsert` then reuses the cached `Table` without making any catalog-level call. `RESTSessionCatalog.close()` closes the root client, shared IO, and tracked table IO, so after (for example) `listNamespaces` heals the catalog, the next scan or INSERT can still receive the cached old object and fail on closed resources until cache refresh/eviction. A catalog call already inside the old delegate can be interrupted the same way and fail with a non-401 that is not retried. Please invalidate/reload the table and view entries when publishing a replacement and keep the retired delegate alive until cached/in-flight users have drained; add a cached-table commit/scan test plus a paused concurrent-call test. -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
