JingsongLi commented on code in PR #9285:
URL: https://github.com/apache/paimon/pull/9285#discussion_r3818041883
##########
paimon-core/src/test/java/org/apache/paimon/table/AppendOnlySimpleTableTest.java:
##########
@@ -325,57 +326,40 @@ public void testDiscardDuplicateFilesMultiThread() throws
Exception {
options.set(CoreOptions.COMMIT_DISCARD_DUPLICATE_FILES, true);
options.set(CoreOptions.COMMIT_MAX_RETRIES, 50);
options.set(CoreOptions.COMMIT_MAX_RETRY_WAIT,
Duration.ofMillis(100));
- // Keep all snapshots so concurrent expiry does
not race readers.
- options.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MIN,
1000);
- options.set(CoreOptions.SNAPSHOT_NUM_RETAINED_MAX,
1000);
});
BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
- List<List<CommitMessage>> messages = new ArrayList<>();
- for (int i = 0; i < 10; i++) {
- try (BatchTableWrite write = writeBuilder.newWrite()) {
- write.write(rowData(1, 10, 100L));
- messages.add(write.prepareCommit());
- }
+ List<CommitMessage> messages;
+ try (BatchTableWrite write = writeBuilder.newWrite()) {
+ write.write(rowData(1, 10, 100L));
+ messages = write.prepareCommit();
}
- int commitThreadNum = 10;
- int commitsPerThread = 10;
- Runnable asserter =
- () -> {
- List<Split> splits =
table.newReadBuilder().newScan().plan().splits();
- assertThat(splits.size()).isEqualTo(1);
- assertThat(splits.get(0).convertToRawFiles().get().size())
- .isLessThanOrEqualTo(messages.size());
- };
+ int commitThreadNum = 5;
ExecutorService pool = Executors.newFixedThreadPool(commitThreadNum);
+ List<Future<?>> futures = new ArrayList<>();
try {
- List<Future<?>> futures = new ArrayList<>();
- for (int thread = 0; thread < commitThreadNum; thread++) {
- int threadId = thread;
+ for (int i = 0; i < commitThreadNum; i++) {
futures.add(
pool.submit(
Review Comment:
[P2] Synchronize the worker starts before committing. Each worker now
performs only one very short task, and a fixed thread pool does not guarantee
that these `commit` calls overlap: on a fast or loaded executor one worker may
finish before another starts. In that execution this becomes the same
sequential duplicate-commit coverage already provided by
`testDiscardDuplicateFiles`, so a concurrency regression can pass unnoticed.
Please use a barrier/latch (all workers signal ready, then are released
together) around `commit(messages)` so the rewritten test reliably reaches the
conflict/retry path.
--
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]
