github-actions[bot] commented on code in PR #66914:
URL: https://github.com/apache/doris/pull/66914#discussion_r3834849651
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java:
##########
@@ -173,24 +174,41 @@ protected void onFail(String errMsg) throws JobException {
@Override
public void cancel(boolean needWaitCancelComplete) {
super.cancel(needWaitCancelComplete);
- if (null != stmtExecutor) {
+ StmtExecutor executor = stmtExecutor;
+ if (null != executor) {
log.info("cancelling streaming insert task, job id is {}, task id
is {}",
getJobId(), getTaskId());
- stmtExecutor.cancel(new Status(TStatusCode.CANCELLED, "streaming
insert task cancelled"),
+ executor.cancel(new Status(TStatusCode.CANCELLED, "streaming
insert task cancelled"),
needWaitCancelComplete);
}
+ if (needWaitCancelComplete) {
+ awaitExecutionCompletion();
Review Comment:
[P1] Do not wait for completion from this task's own worker. A terminal
failure runs on the scheduler worker through
execute()->onFail()->onStreamTaskFail()->updateJobStatus(PAUSED)->cancelAllTasks(true)->cancel(true),
so this line waits for executionFinished even though that flag is set only by
execute()'s outer finally after onFail returns. The failure path therefore
self-deadlocks before the PAUSE transition can finish. External PAUSE also
reaches this wait while holding the job write lock, which can block a worker
entering beforeCommitted on that same lock. Move cancellation/waiting outside
the job lock and make the completion handoff detect or avoid owner-thread
waits; add latch tests for both failure-driven and external PAUSE paths.
##########
fe/fe-connector/fe-connector-iceberg/src/main/java/org/apache/doris/connector/iceberg/IcebergConnector.java:
##########
@@ -896,13 +902,88 @@ boolean isUserSessionEnabled() {
&&
IcebergCatalogProperties.TYPE_REST.equals(catalogProps.getFlavor());
}
- private Catalog getOrCreateCatalog() {
- if (icebergCatalog == null) {
- synchronized (this) {
- if (icebergCatalog == null) {
- icebergCatalog = createCatalog();
- }
+ /**
+ * Closes a table's FileIO when the cached raw table owns it. Glue and
S3Tables create a per-table
+ * S3FileIO; REST tables are closed only when they do not share the
catalog-level FileIO. Other catalog
+ * flavors are left untouched because they may share a catalog-level
FileIO.
+ */
+ private Runnable cachedTableCleanup(Table table) {
+ return cachedTableCleanup(table, catalogProps.getFlavor(),
restSessionCatalog);
+ }
+
+ static Runnable cachedTableCleanup(Table table, String flavor) {
+ return cachedTableCleanup(table, flavor, null);
+ }
+
+ private static Runnable cachedTableCleanup(Table table, String flavor,
Object catalog) {
+ if (table == null) {
+ return () -> { };
+ }
+ boolean tableOwned = false;
+ try {
+ if (IcebergCatalogProperties.TYPE_GLUE.equals(flavor)
+ || IcebergCatalogProperties.TYPE_S3_TABLES.equals(flavor))
{
+ tableOwned = true;
+ } else if (IcebergCatalogProperties.TYPE_REST.equals(flavor)) {
+ FileIO catalogFileIO = restCatalogFileIO(catalog);
Review Comment:
[P1] Bind REST FileIO ownership to the delegate generation that produced
this table. A cache miss can return a table from delegate G, then a concurrent
catalog-identity 401 can publish G+1 before this cleanup factory runs.
restCatalogFileIO() unwraps currentDelegate(), so G's shared catalog IO is
compared with G+1's IO and marked table-owned. Retiring this owner then
directly closes G.io while other G borrowers still hold generation leases;
those leases defer delegate.close() but cannot undo the direct FileIO close.
Capture the producing generation/catalog IO with the load and add a latch test
for rotation after loadTable returns but before cleanup classification.
##########
fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectPoolMgr.java:
##########
@@ -77,7 +77,7 @@ public void unregisterConnection(ConnectContext ctx) {
// Finalize any Arrow Flight query whose coordinator was kept alive
across the
// GetFlightInfo -> DoGet phases (see #62259), releasing its resources
(e.g. external-table
// batch SplitSources and the query queue slot).
- ctx.closeFlightSqlDeferredExecutors();
+ ctx.sealAndCloseFlightSqlDeferredExecutors();
Review Comment:
[P1] Publish the terminal seal before destroying the Flight result channel.
A local-result GetFlightInfo can pause just before publishFlightInfo();
unregisterConnection then closes the channel, invalidating its cached result,
but has not set the seal until this line. The publisher can resume in that
window, end with sealed=false, and return a successful ticket whose DoGet
result is already gone. Move sealAndCloseFlightSqlDeferredExecutors() ahead of
flightSqlChannel.close(), and add a latch test that proves teardown cannot
close the local result before publication either fails or commits.
--
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]
