Copilot commented on code in PR #41:
URL: https://github.com/apache/pulsar-connectors/pull/41#discussion_r3554092094
##########
file/src/test/java/org/apache/pulsar/io/file/ProcessedFileThreadTest.java:
##########
@@ -41,6 +41,22 @@ public class ProcessedFileThreadTest extends
AbstractFileTest {
private ProcessedFileThread cleanupThread;
private FileSourceConfig fileConfig;
+ /**
+ * Waits (bounded) until every produced file has made it through the
entire pipeline,
+ * including the final rename/delete performed by the cleanup thread.
Checking only the
+ * queues is not enough: a file is briefly in none of them while it is
handed from one
+ * thread to the next, and the last hop (disk rename/delete) happens after
the file has
+ * already left the queues.
+ */
+ private void awaitProcessingComplete() throws InterruptedException {
+ long deadline = System.currentTimeMillis() + 60_000;
+ while (System.currentTimeMillis() < deadline
+ && (!workQueue.isEmpty() || !inProcess.isEmpty() ||
!recentlyProcessed.isEmpty()
+ || producedFiles.stream().anyMatch(File::exists))) {
+ Thread.sleep(200);
+ }
+ }
Review Comment:
`awaitProcessingComplete()` returns silently after the 60s deadline even if
the pipeline still hasn’t drained. In
`continuousRunTest`/`multipleConsumerTest` that can allow the test to continue
and potentially pass without actually waiting for the final delete/rename on
disk (which this helper’s Javadoc says it is supposed to guarantee). Make the
helper fail when the timeout is hit and the completion condition is still not
satisfied so the tests can’t produce false positives.
--
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]
