Spring Integration: File polling memory consumption

Question

I've following InboundChannelAdapter with Poller to process files every 30 seconds. The files are not large but I realize the memory consumptions keeps going up even when there's no files coming.

@Bean
@InboundChannelAdapter(value = "flowFileInChannel" ,poller = @Poller(fixedDelay ="30000", maxMessagesPerPoll = "1"))
public MessageSource<File> flowInboundFileAdapter(@Value("${integration.path}") File directory) {
    FileReadingMessageSource source = new FileReadingMessageSource();
    source.setDirectory(directory);
    source.setFilter(flowPathFileFilter);
    source.setUseWatchService(true);
    source.setScanEachPoll(true);
    source.setAutoCreateDirectory(false);
    return source;
}

Is there an internal queue that is not cleared after each poll? How do I configure to avoid eating up memory.

After digging deeper, it looks like the below Spring IntegrationFlows which processes the data from the InboundChannelDapter is holding up the memory after each file polling. After I commenting out the middle part, the memory consumption seems stable (instead of increasing consumption). Now I'm wondering how do we force Spring IntegrationFlows to clear those Messages and Headers after they're passed through different channels (i.e. after the last channel below)

public IntegrationFlow incomingLocateFlow(){
        return IntegrationFlows.from(locateIncomingChannel())

//                .split("locateItemSplitter","split")
//                .transform(locateItemEnrichmentTransformer)
//                .transform(locateRequestTransformer)
//                .aggregate(new Consumer<AggregatorSpec>() {                        // 32
//
//                    @Override
//                    public void accept(AggregatorSpec aggregatorSpec) {
//                        aggregatorSpec.processor(locateRequestProcessor, null);                // 33
//                    }
//
//                }, null)
//                .transform(locateIncomingResultTransformer)
//                .transform(locateExceptionReportWritingHandler)
                .channel(locateIncomingCompleteChannel())
                .get();
    }

Artem Bilan Artem Bilan · Accepted Answer · 2017-07-11T13:28:18

Indeed there is an AcceptOnceFileListFilter with the code like:

private final Queue<F> seen;

private final Set<F> seenSet = new HashSet<F>();

On each poll those internal collections are replenished with new files.

For this purpose you can consider to use FileSystemPersistentAcceptOnceFileListFilter with the persistent MetadataStore implementation to avoid memory consumption.

Also consider to use some tool to analyze the memory content. You might have something else downstream on the flowFileInChannel.

UPDATE

Since you use .aggregate() it is definitely the point where memory is consumed by default. That's because there is SimpleMessageStore to keep messages for grouping. Plus there is an option expireGroupsUponCompletion(boolean) which is false by default. Therefore even after successful releasing some info is still in the MessageStore. That's how your memory is consumed a bit from time to time.

That option is false by default to let to have logic when we discard late message for completed group. When it is true, you are able to form fresh group for the same correlationKey.

See more info about Aggregator in the Reference Manual.

Spring Integration: File polling memory consumption

1 Answers