INT-4560: Fix Race in FileSystemPersistentAOFLF
JIRA: https://jira.spring.io/browse/INT-4560 Reproduced and tested with ```java @SpringBootApplication public class So53521593Application { private static final Logger logger = LoggerFactory.getLogger(So53521593Application.class); public static void main(String[] args) { SpringApplication.run(So53521593Application.class, args); } @Bean public IntegrationFlow flow() { ExecutorService exec = Executors.newFixedThreadPool(10); return IntegrationFlows.from(Files.inboundAdapter(new File("/tmp/foo")).filter( new MyFilter(new SimpleMetadataStore(), "foo")), e -> e.poller(Pollers.fixedDelay(5, TimeUnit.SECONDS) .maxMessagesPerPoll(10))) .channel(MessageChannels.executor(exec)) .<File>handle((p, h) -> { try { p.delete(); logger.info(p.toString()); Thread.sleep(10_000); } catch (InterruptedException e1) { Thread.currentThread().interrupt(); } return null; }) .get(); } } class MyFilter extends FileSystemPersistentAcceptOnceFileListFilter { public MyFilter(ConcurrentMetadataStore store, String prefix) { super(store, prefix); } @Override protected long modified(File file) { long modified = super.modified(file); System.out.println(modified); return modified; } } ``` **cherry-pick to 5.0.x, 4.3.x**
This commit is contained in:
committed by
Artem Bilan
parent
a4746f79ae
commit
39fef2c70a
@@ -77,17 +77,27 @@ public abstract class AbstractPersistentAcceptOnceFileListFilter<F> extends Abst
|
||||
String oldValue = this.store.putIfAbsent(key, newValue);
|
||||
if (oldValue == null) { // not in store
|
||||
flushIfNeeded();
|
||||
return true;
|
||||
return fileStillExists(file);
|
||||
}
|
||||
// same value in store
|
||||
if (!isEqual(file, oldValue) && this.store.replace(key, oldValue, newValue)) {
|
||||
flushIfNeeded();
|
||||
return true;
|
||||
return fileStillExists(file);
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if the file still exists; default implementation returns true.
|
||||
* @param file the file.
|
||||
* @return true if the filter should return true.
|
||||
* @since 4.3.19
|
||||
*/
|
||||
protected boolean fileStillExists(F file) {
|
||||
return true;
|
||||
}
|
||||
|
||||
/**
|
||||
* {@inheritDoc}
|
||||
* @since 4.0.4
|
||||
|
||||
@@ -42,4 +42,15 @@ public class FileSystemPersistentAcceptOnceFileListFilter extends AbstractPersis
|
||||
return file.getAbsolutePath();
|
||||
}
|
||||
|
||||
/**
|
||||
* Check that the file still exists, to avoid a race condition when multi-threaded and
|
||||
* another thread removed the file while we were waiting for the lock.
|
||||
* @since 4.3.19
|
||||
*/
|
||||
@Override
|
||||
protected boolean fileStillExists(File file) {
|
||||
return file.exists();
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user