JingsongLi commented on code in PR #10133:
URL: https://github.com/apache/paimon/pull/10133#discussion_r4253899151
##########
paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreWrite.java:
##########
@@ -703,12 +713,98 @@ private void checkNumBuckets(String partInfo, int
expected, int previous) {
}
}
- private ExecutorService compactExecutor() {
+ private ExecutorService compactExecutor(BinaryRow partition, int bucket) {
+ if (externalCompactExecutor) {
+ return lazyCompactExecutor;
+ }
+
+ switch (compactionTaskExecutorMode) {
+ case PER_BUCKET:
+ return perBucketCompactExecutors.computeIfAbsent(
+ new BucketCompactionExecutorKey(partition, bucket),
+ key ->
+ Executors.newSingleThreadExecutor(
+ new ExecutorThreadFactory(
+
Thread.currentThread().getName()
+ + "-compaction-bucket-"
+ + key.bucket)));
+ case FIXED_POOL:
+ return sharedCompactionExecutor(compactionTaskThreads,
"-compaction-pool");
+ case SINGLE:
+ default:
+ return sharedCompactionExecutor(1, "-compaction");
+ }
+ }
+
+ private void releaseCompactionExecutor(BinaryRow partition, int bucket) {
+ if (compactionTaskExecutorMode != CompactionTaskExecutorMode.PER_BUCKET
+ || externalCompactExecutor) {
Review Comment:
[P2] Clean up owned bucket executors after a late external override
In the CDC schema-refresh path, StoreSinkWriteImpl.replace() restores active
buckets before the next processElement() installs the external executor. With
compaction.task-threads=-1, those restored writers retain internal dedicated
executors. Setting externalCompactExecutor=true changes routing for subsequent
buckets, but the guard here also skips releasing the already-owned executors
when their restored writers become idle. Their worker threads, map entries and
bucket timers remain until another replacement or full close. A focused
regression starts an internal worker, installs an external executor, releases
the old bucket and observes one retained internal executor instead of zero; the
external executor remains correctly routed to new buckets. Please decide
cleanup from actual per-bucket ownership (remove/shut down an owned map entry
and retire only its timer), rather than the global external flag, and preserve
the external executor before restoration where appropriate.
--
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]
