GH-8967: maxFetchSize = 1 from the StandardRotationPolicy
Fixes: #8967 In the `fair` mode the `StandardRotationPolicy` re-configures an `AbstractFetchLimitingMessageSource` for a new directory (and possible new `ConnectionFactory`) in the `beforeReceive()`. However, with default `maxFetchSize` (or bigger than `1`), the `receive()`` would poll `toBeReceived` internal queue for files cached from the previous polling cycle. Since we rotate the source immediately to a new set of options, all those cached files make no sense or even can cause the problem on fetching their content in case of `AbstractRemoteFileStreamingMessageSource` when we rotate to a new `ConnectionFactory`. * Call `fetchLimitingMessageSource.setMaxFetchSize(1);` in the `StandardRotationPolicy.beforeReceive()` when `fair && !this.initialized` **Auto-cherry-pick to `6.1.x`**
This commit is contained in:
@@ -1,5 +1,5 @@
|
||||
/*
|
||||
* Copyright 2018-2023 the original author or authors.
|
||||
* Copyright 2018-2024 the original author or authors.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
@@ -405,8 +405,7 @@ public class RotatingServersTests extends FtpTestSupport {
|
||||
public IntegrationFlow flow() {
|
||||
return IntegrationFlow.from(Ftp.inboundStreamingAdapter(new FtpRemoteFileTemplate(sf()))
|
||||
.filter(new FtpPersistentAcceptOnceFileListFilter(new SimpleMetadataStore(), "rotate"))
|
||||
.remoteDirectory(".")
|
||||
.maxFetchSize(1),
|
||||
.remoteDirectory("."),
|
||||
e -> e.poller(Pollers.fixedDelay(1).advice(advice())))
|
||||
.channel(MessageChannels.queue("files"))
|
||||
.get();
|
||||
|
||||
Reference in New Issue
Block a user