924060929 commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3900832347
##########
fe/fe-core/src/main/java/org/apache/doris/datasource/hudi/source/HudiScanNode.java:
##########
@@ -494,22 +543,52 @@ private List<HudiSplit> planPartitionSplits(HivePartition
partition) throws IOEx
private void getPartitionsSplits(List<HivePartition> partitions,
List<Split> splits) {
Executor executor =
Env.getCurrentEnv().getExtMetaCacheMgr().getFileListingExecutor();
- CountDownLatch countDownLatch = new CountDownLatch(partitions.size());
+ ListingFsViewOwner createdOwner = new ListingFsViewOwner(fsViewLease);
+ ListingFsViewOwner owner = createdOwner;
+ ConnectContext connectContext = ConnectContext.get();
+ StatementContext statementContext = connectContext == null ? null :
connectContext.getStatementContext();
+ if (statementContext != null) {
+ try {
+ owner =
statementContext.getOrRegisterStatementResource(listingFsViewResourceKey, () ->
createdOwner);
+ if (owner != createdOwner) {
+ throw new IllegalStateException("Hudi listing owner was
registered twice");
+ }
+ } catch (RuntimeException e) {
+ createdOwner.discardBeforeSubmission();
+ throw e;
+ }
+ }
+ // The owner now releases the exact fs-view generation after every
accepted task terminates.
+ if (!fsViewReleased.compareAndSet(false, true)) {
+ owner.discardBeforeSubmission();
+ throw new IllegalStateException("Hudi filesystem-view lease has
already been released");
+ }
AtomicReference<Throwable> throwable = new AtomicReference<>();
+ RuntimeException submissionFailure = null;
long startTime = System.currentTimeMillis();
- partitions.forEach(partition -> executor.execute(() -> {
+ for (HivePartition partition : partitions) {
Review Comment:
Fixed in ad72e8b65f2. ListingFsViewOwner now owns task registration and
submission, stops the partition loop once statement cleanup begins, and removes
cancelled queued tasks. The bounded BlockedPolicy now observes cancelled Future
tasks while waiting, so a close during a full-queue submission exits promptly
without interrupting the planner thread or losing an external interrupt. Added
deterministic tests for close during blocked submission, queued-task removal,
immediate close, cancelled Future rejection, and interrupt preservation.
Focused FE UT (27 tests), Checkstyle, and ./build.sh --fe all pass.
--
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]
