Nio locker usage in spring integration java dsl for file polling

Question

I'm trying to implement a spring integration file polling where multiple servers independently read from a common directory and process the file. Each of them after processing the file rename it .DONE so the other doesn't pick it. After observing some files being picked by both I have used nioLocker as in the sample below. But looks like randomly this doesn't work. Let me know the necessary steps to make nioLocker reliable in a multi-server environment. For other reasons having a seperate mediator (like zookeepr/mongo etc) is not an option.

Thanks.

 @Bean
    public TransactionSynchronizationFactory transactionSynchronizationFactory() {
        ExpressionParser parser = new SpelExpressionParser();
        ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor =
                new ExpressionEvaluatingTransactionSynchronizationProcessor();
        syncProcessor.setAfterCommitExpression(parser.parseExpression("payload.renameTo(new java.io.File(payload.path+'.DONE'))"));
        syncProcessor.setAfterRollbackExpression(parser.parseExpression("payload.renameTo(new java.io.File(payload.path+'.DONE'))"));
        return new DefaultTransactionSynchronizationFactory(syncProcessor);
    }

@Bean
public IntegrationFlow receiveInputFile(@Value("/opt/tomcat/in/test") File in,
                                            @Value(".txt") String pattern,
                                            @Value("${edi.poll.delay.all.edi}") int delay,
                                            @Value("${edi.messages.per.poll.new.order}") int messagesPerPoll) {

    //Logging required config for debugging
    LOGGER.debug("EDI File pattern :"+pattern);
    LOGGER.debug("EDI Delay Seconds :"+delay);
    LOGGER.debug("EDI Messages Per Poll :"+messagesPerPoll);


    return IntegrationFlows
            .from(s -> s.file(in).patternFilter(ServicesConstant.PATTERN_PREFIX + pattern).scanEachPoll(true).nioLocker(),
                    e -> e.poller(Pollers.fixedDelay(delay).maxMessagesPerPoll(messagesPerPoll)
                            .transactionSynchronizationFactory(transactionSynchronizationFactory())
                            .transactional(new PseudoTransactionManager())))
            .handle(m -> {
                LOGGER.info("Received test file " + m);
                LOGGER.info("File path: " +m.getPayload());
                LOGGER.info(""+ m.getHeaders().toString());
            })
            .get();
}

Artem Bilan Artem Bilan · Accepted Answer · 2018-07-24T20:45:14

See its JavaDocs:

/**
 * File locking strategy that uses java.nio. The locks taken by FileChannel are shared with all the threads in a single
 * JVM, so this locking strategy <b>does not</b> prevent files being picked up multiple times within the same JVM.
 * {@link FileReadingMessageSource}s sharing a Locker will not pick up the same files.
 * <p>
 * This implementation will acquire or create a {@link FileLock} for the given file. Caching locks might be expensive,
 * so this locking strategy is not recommended for scenarios where many files are accessed in parallel.
 *
 * @author Iwein Fuld
 * @author Mark Fisher
 * @since 2.0
 */
public class NioFileLocker extends AbstractFileLockerFilter {

In my practice we really have a good exclusive file locking only on Windows.

Saying that I'm not sure what "a multi-server environment" means for you, if we deal only with a single local file system. I'm pretty sure that this NIO locking does't have effect for the shared network directory.

If you can't use any shared persistent store for file filtering, I'd suggest you to take a look into the FileSystemPersistentAcceptOnceFileListFilter with really some shared MetadataStore implementation to let all your server to negotiate with it to be sure that file has not been processed yet by some other server.

Nio locker usage in spring integration java dsl for file polling

1 Answers