github-actions[bot] commented on code in PR #66914:
URL: https://github.com/apache/doris/pull/66914#discussion_r3821703333
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java:
##########
@@ -271,7 +273,8 @@ public IcebergConnector(Map<String, String> properties,
ConnectorContext context
||
IcebergScanPlanProvider.restVendedCredentialsEnabled(this.properties))
? null
: new IcebergTableCache(
- resolveTableCacheTtlSecond(this.properties),
DEFAULT_TABLE_CACHE_CAPACITY);
+ resolveTableCacheTtlSecond(this.properties),
DEFAULT_TABLE_CACHE_CAPACITY,
+ this::cachedTableCleanup, catalogResourceTracker);
Review Comment:
[P1] Enroll mutable operations in catalog generations instead of wiring the
tracker only to the read cache. This constructor passes
`catalogResourceTracker` only to `IcebergTableCache`, but write planning and
`beginWrite` use `sharedWritableTable`, which memoizes a raw `Table`; the SDK
transaction then retains its `TableOperations` through FE commit. If an
unrelated catalog-identity 401 rotates the REST delegate while that write is
active, the tracker sees no writer reference and can close the old REST
client/FileIO tracker before `commitTransaction()` uses it. Procedure/DDL table
and builder paths bypass `tableCache` through the same seam, so the repair must
cover shorter operation-owned objects too. Transfer the write lease to the
transaction through commit/rollback/close, retain other mutable operations to
completion, and test active writes/mutations across both 401 rotation and
connector reset.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/ReauthenticatingRestSessionCatalog.java:
##########
@@ -124,9 +134,22 @@ private synchronized void
reauthenticate(RESTSessionCatalog attemptedOn, Runtime
+ "then retrying the request once.", name(), cause);
RESTSessionCatalog replacement = delegateBuilder.get();
RESTSessionCatalog wedged = delegate;
- delegate = replacement;
+ if (resourceTracker == null) {
+ delegate = replacement;
+ closeReplacedDelegate(wedged);
+ return;
+ }
+ try {
+ invalidateTables.run();
+ } catch (RuntimeException e) {
+ LOG.warn("Failed to retire Iceberg table cache before replacing
REST client of catalog {}", name(), e);
+ }
+ resourceTracker.rotate(() -> closeReplacedDelegate(wedged), () ->
delegate = replacement);
Review Comment:
[P2] Close an unpublished replacement when teardown wins this race.
`delegateBuilder.get()` creates a fully initialized REST catalog before
`rotate()` checks whether the tracker was closed. A concurrent connector reset
can set `closed` in that interval, so `rotate()` throws, the replacement is
never assigned to `delegate` or closed, and the original 401 is replaced by an
internal `IllegalStateException`. Please coordinate close with replacement
acceptance or close the rejected replacement in `finally` while preserving the
primary outcome, and add a latch-driven close/build/rotate test.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergScanPlanProvider.java:
##########
@@ -3019,16 +3020,24 @@ private Table resolveTable(ConnectorSession session,
IcebergTableHandle handle)
// Resolve the per-request ops before the auth scope so a session=user
fail-closed surfaces verbatim (it
// re-validates the credential even on a scope hit).
IcebergCatalogOps ops = catalogOpsResolver.apply(session);
- Table raw = IcebergStatementScope.sharedTable(session,
handle.getDbName(), handle.getTableName(), () -> {
+ Supplier<Table> directLoader = () -> {
try {
- return context == null
- ? loadRawTable(ops, handle)
- : context.executeAuthenticated(() -> loadRawTable(ops,
handle));
+ return context == null ? ops.loadTable(handle.getDbName(),
handle.getTableName())
+ : context.executeAuthenticated(
+ () -> ops.loadTable(handle.getDbName(),
handle.getTableName()));
} catch (Exception e) {
throw IcebergExceptionUtils.wrapTableLoadFailure(
handle, e, "Failed to load table for scan, error
message is:");
}
- });
+ };
+ Table raw = tableCache == null
Review Comment:
[P1] Keep a generation lease when the cross-query cache is disabled. For
session-user and REST vended-credential catalogs, `tableCache` is null, so this
branch stores a raw `Table` in the statement scope and never calls
`catalogResourceTracker.beginLoad()`. If connector reset or a concurrent
catalog-identity 401 retires the delegate, the tracker sees no reference and
Iceberg closes the delegate's `FileIOTracker`, including the active table's
per-table IO; later planning can then use a closed resource. This is distinct
from the existing cached-borrower thread because these authorization-required
paths cannot create a `TableOwner`/`TableLease` at all. Please give direct
statement-owned tables a generation lease without cross-query caching, and test
connector close plus catalog-identity rotation for both configurations.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergCatalogResourceTracker.java:
##########
@@ -0,0 +1,196 @@
+// 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.connector.iceberg;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/** Keeps catalog generations alive while cached tables loaded through them
still have borrowers. */
+final class IcebergCatalogResourceTracker {
+
+ private final List<Generation> generations = new ArrayList<>();
+ private Generation current = new Generation();
+ private int loadsInProgress;
+ private boolean closed;
+
+ IcebergCatalogResourceTracker() {
+ generations.add(current);
+ }
+
+ synchronized LoadGuard beginLoad() {
+ if (closed) {
+ throw new IllegalStateException("Iceberg catalog resources are
already closed");
+ }
+ loadsInProgress++;
+ current.retain();
+ return new LoadGuard(this, generations.size() - 1, current);
+ }
+
+ /** Atomically rotates the resource generation with publication of the
corresponding REST delegate. */
+ synchronized void rotate(Runnable retiredCleanup, Runnable
publishReplacement) {
+ if (closed) {
+ throw new IllegalStateException("Iceberg catalog resources are
already closed");
+ }
+ Generation retired = current;
+ current = new Generation();
+ current.retain(loadsInProgress);
+ generations.add(current);
+ publishReplacement.run();
+ retired.retire(retiredCleanup);
+ }
+
+ synchronized void close(Runnable currentCleanup) {
+ if (closed) {
+ return;
+ }
+ closed = true;
+ current.retire(currentCleanup);
+ }
+
+ private synchronized ResourceLease promote(int firstGeneration) {
+ List<Generation> retained = new ArrayList<>();
+ retained.add(generations.get(firstGeneration));
+ for (int i = firstGeneration + 1; i < generations.size(); i++) {
+ // rotate() pre-retains one reference in every new generation for
each load already in progress.
+ retained.add(generations.get(i));
+ }
+ completeLoad();
+ return new ResourceLease(retained);
+ }
+
+ private synchronized void abortLoad(int firstGeneration) {
+ for (int i = firstGeneration + 1; i < generations.size(); i++) {
+ generations.get(i).release();
+ }
+ completeLoad();
+ }
+
+ private void completeLoad() {
+ loadsInProgress--;
+ if (loadsInProgress < 0) {
+ throw new IllegalStateException("Iceberg catalog load guard
completed too many times");
+ }
+ }
+
+ static final class LoadGuard implements AutoCloseable {
+ private final IcebergCatalogResourceTracker tracker;
+ private final int firstGeneration;
+ private final Generation initialGeneration;
+ private final AtomicBoolean transferred = new AtomicBoolean();
+
+ private LoadGuard(IcebergCatalogResourceTracker tracker, int
firstGeneration, Generation initialGeneration) {
+ this.tracker = tracker;
+ this.firstGeneration = firstGeneration;
+ this.initialGeneration = initialGeneration;
+ }
+
+ ResourceLease promote() {
+ if (!transferred.compareAndSet(false, true)) {
+ throw new IllegalStateException("Iceberg catalog load guard
was already completed");
+ }
+ return tracker.promote(firstGeneration);
+ }
+
+ @Override
+ public void close() {
+ if (transferred.compareAndSet(false, true)) {
+ try {
+ initialGeneration.release();
+ } finally {
+ tracker.abortLoad(firstGeneration);
+ }
+ }
+ }
+ }
+
+ static final class ResourceLease implements AutoCloseable {
+ private final List<Generation> generations;
+ private final AtomicBoolean closed = new AtomicBoolean();
+
+ private ResourceLease(List<Generation> generations) {
+ this.generations = generations;
+ }
+
+ @Override
+ public void close() {
+ if (closed.compareAndSet(false, true)) {
+ RuntimeException failure = null;
+ for (Generation generation : generations) {
+ try {
+ generation.release();
+ } catch (RuntimeException e) {
+ if (failure == null) {
+ failure = e;
+ } else {
+ failure.addSuppressed(e);
+ }
+ }
+ }
+ if (failure != null) {
+ throw failure;
+ }
+ }
+ }
+ }
+
+ private static final class Generation {
+ private int references;
+ private boolean retired;
+ private boolean cleaned;
+ private Runnable cleanup;
+
+ private synchronized void retain() {
+ if (cleaned) {
+ throw new IllegalStateException("Iceberg catalog generation
was already cleaned");
+ }
+ references++;
+ }
+
+ private synchronized void retain(int count) {
+ if (cleaned) {
+ throw new IllegalStateException("Iceberg catalog generation
was already cleaned");
+ }
+ references += count;
+ }
+
+ private synchronized void retire(Runnable cleanup) {
+ if (retired) {
+ return;
+ }
+ retired = true;
+ this.cleanup = cleanup;
+ maybeCleanup();
+ }
+
+ private synchronized void release() {
+ if (references <= 0) {
+ throw new IllegalStateException("Iceberg catalog generation
released too many times");
+ }
+ references--;
+ maybeCleanup();
+ }
+
+ private void maybeCleanup() {
+ if (retired && references == 0 && !cleaned) {
+ cleaned = true;
+ cleanup.run();
Review Comment:
[P2] Release the retired delegate after its cleanup runs. `generations`
permanently retains every `Generation`, and the cleanup stored by `retire()`
captures the wedged `RESTSessionCatalog`; `cleanup.run()` marks it closed but
leaves that lambda reachable forever. Repeated 401 recoveries therefore retain
every old client/auth/FileIO object graph for the connector's lifetime. Clear
the cleanup in a `finally` (and preferably prune cleaned generations without
index-based guards), with repeated-rotation coverage that proves retired
delegates are no longer retained.
--
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]
