Fixes: #8879
To work around the problem with `$share/` subscriptions the `Mqttv5PahoMessageDrivenChannelAdapter` must provide a `subscriptionIdentifier` into `MqttProperties` on `subscribe()`
* Introduce a `Mqttv5PahoMessageDrivenChannelAdapter.subscriptionIdentifierCounter` according to the MQTT specification:
> 3.8.2.1.2 Subscription Identifier: [..]The Subscription Identifier is associated with any subscription created or modified as the result of this SUBSCRIBE packet. If there is a Subscription Identifier, it is stored with the subscription.
This one is associated with the MQTT session for the current subscriber and does not interfere into other sessions even if identifier is same from the counter.
It works because the Subscription identifier is per session and because you cannot have multiple connection with the same client ID.
**Cherry-pick to `6.2.x` & `6.1.x`**
# Conflicts:
# spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java
Fixes: #8873
The `mqttClient.subscribe()` API does not check if properties are provided and fails on the
`subscriptionProperties.getSubscriptionIdentifiers().get(0)` call with an `IndexOutOfBoundsException`
* Use another `mqttClient.subscribe()` API in the `Mqttv5PahoMessageDrivenChannelAdapter` where there is not such a check
* Ensure that `addTopic(NAME)` works as expected in the `Mqttv5BackToBackTests`
**Cherry-pick to `6.2.x` & `6.1.x`**
# Conflicts:
# spring-integration-mqtt/src/main/java/org/springframework/integration/mqtt/inbound/Mqttv5PahoMessageDrivenChannelAdapter.java
Fixes: #8869
Do not set the `redisMessageListenerContainer`'s executor in the setter method
since the Redis Message Listener container is not initialised during configuration due to lazy loading
**Cherry-pick to `6.2.x` & `6.0.x`**
(cherry picked from commit 62fd3e6c20)
Fixes: #8852
The `while (true) {` causes the thread to be blocked forever.
The `stop()` waits on the `if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {` the whole time
* Fix with the `while (isActive()) {`
* Remove the `ConnectionFactoryTests.testEarlyCloseNet()` since we cannot reach expectations because of
race condition between `start()` and `stop()`
**Cherry-pick to `6.2.x` & `6.1.x`**
(cherry picked from commit 0b60dfc9d1)
Fixes: #8803
Would be great to have an `Acquired lock; RedisLock...` logging message as we already have `Released lock; RedisLock ...`
* Improve `RedisLockRegistry.RedisLock.tryRedisLock()` to emit `Acquired lock; RedisLock...` debug message
**Cherry-pick to `6.0.x`**
(cherry picked from commit 9bd8001a61)
The Checkstyle violation for unused imports in the `FileTailingMessageProducerTests`
Leftover from the previous fix for this class
**Cherry-pick to `6.1.x`**
(cherry picked from commit a6f826e1cf)
The `OSDelegatingFileTailingMessageProducer` publishes event from scheduled task
* Modify `FileTailingMessageProducerTests.testGuts()` to use `CountDownLatch`
to wait for those asynchronous events
**Cherry-pick to `6.1.x`**
(cherry picked from commit a3d347eddd)
Fixes: gh-8800
The `SmbFile.getName()` returns plain file name, not with a
sub-directory as it is expected by the `AbstractRemoteFileOutboundGateway`
for (S)FTP protocols
* Introduce `AbstractRemoteFileOutboundGateway.getFullFileName()`
with dir and file name concatenation by default (as it was before)
and override it in the `SmbOutboundGateway` to return the full path according this protocol.
* Fix `SmbOutboundGateway.enhanceNameWithSubDirectory()` to return an `SmbFile` instance
since it has all the info about remote file, include requested directory
* Fix `SmbOutboundGateway.getFilename()` to remove the trailing `/` in the dir name
since this is not what expected by the `AbstractRemoteFileOutboundGateway` logic
**Cherry-pick to `6.1.x`**
* Remove JFrog plugin since it is supplied from CI/CD by Gradle init script
* Remove Sonar and Jacoco since we don't manage those service anymore
# Conflicts:
# .github/release-files-spec.json
# .github/workflows/central-sync-close.yml
# .github/workflows/central-sync-create.yml
# .github/workflows/central-sync-release.yml
# .github/workflows/ci-snapshot.yml
# .github/workflows/pr-build-workflow.yml
# build.gradle
Fixes https://github.com/spring-projects/spring-integration/issues/8792
1. Use a `SftpStreamingMessageSource` with a `maxFetchSize = 5` and a `ChainFileListFilter` filter composed with
`SftpSystemMarkerFilePresentFileListFilter` which `supportsSingleFileFiltering == false`
2. Put 2 files in the folder and invoke `SftpStreamingMessageSource.receive()` method twice.
3. Put 5 files in the folder and invoke `SftpStreamingMessageSource.receive()`` method five times.
4. The last two files won't be received.
When you set max fetch size to a number bigger than one (for example 5) and at a certain point
it is necessary to `this.toBeReceived.clear()` inside `AbstractRemoteFileStreamingMessageSource.doReceive()`,
those removed elements from toBeReceived are not rolled back.
* Fix `AbstractRemoteFileStreamingMessageSource.listFiles()` to calculate `maxFetchSize` as `getMaxFetchSize() - this.fetched.get()`
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**
(cherry picked from commit 22c4db291b)
Fixes https://github.com/spring-projects/spring-integration/issues/8786
If `FtpSession.readRaw()` fails, the next `FtpSession.finalizeRaw()` call
would lead to `FTPClient.completePendingCommand()` failure since
there is no command to finish.
* Fix `FtpSession.finalizeRaw()` to exit earlier positively in case of
`FTPReply.isNegativePermanent()` for the current reply code set by the failure
from a previous `FtpSession.readRaw()`
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**
Fixes https://github.com/spring-projects/spring-integration/issues/8785
The `ClientWebSocketContainer.start()` delegates to the `IntegrationWebSocketConnectionManager`
which performs an async connection to the server.
* Wait for `connectionLatch` in the `ClientWebSocketContainer.start()`
and check for `this.openConnectionException != null` to re-throw.
Mark `ClientWebSocketContainer` as stopped in that case
**Cherry-pick to `6.1.x` & `6.0.x`**
# Conflicts:
# spring-integration-websocket/src/main/java/org/springframework/integration/websocket/ClientWebSocketContainer.java
The `this.messagePrefix + groupId + '_' + messageId` logic
is introduced since `6.2` via: 64f8ed5bab
The current fix for new `getMessageFromGroup()` and `removeMessageFromGroupById()`
is to retrieve group and then perform plain `get` and `delete` against just message id
Related to: https://github.com/spring-projects/spring-integration/issues/8773
Fixes https://github.com/spring-projects/spring-integration/issues/8773
The https://github.com/spring-projects/spring-integration/issues/8732 introduced a filtering for messages in group.
So, plain `removeMessage()` doesn't work any more if message is connected to some group yet.
Therefore, `DelayHandler` is failing.
* Introduce `getMessageFromGroup()` and `removeMessageFromGroupById()` into `MessageGroupStore` API
and implement it respectively in all the stores
* Remove `@LongRunningTest` from delayer integration tests and adjust its config to delay not for a long
**Cherry-pick to `6.1.x`**
# Conflicts:
# spring-integration-core/src/main/java/org/springframework/integration/handler/DelayHandler.java
* GH-8778: Fix KafkaMessageSource deadlock
Fixes https://github.com/spring-projects/spring-integration/issues/8778
The `KafkaMessageSource.doReceive()` have a lock around its whole body.
That includes the `pollRecord()` which can be blocked on the `KafkaConsumer.poll()`.
This way the rest of lifecycle management callbacks can be blocked until `KafkaConsumer.poll()` returns.
* Rework lifecycle management flags to `AtomicBoolean` since there is not too much work
in their respective callbacks
* Decrease a locking block in the `doReceive()` just to consumer setup part.
Leave `pollRecord()` outside of the lock
* Add `this.consumer.wakeup()` into `stopConsumer()` to break a `poll()` cycle
and return immediately for the next `close()` call
**Cherry-pick to `6.1.x` & `6.0.x`**
* * Use `compareAndSet` in `start` & `stop`
* GH-8770: Add `PostgresSubsChannel.errorHandler`
Fixes https://github.com/spring-projects/spring-integration/issues/8770
The problem with the `PostgresSubscribableChannel.notifyUpdate()` is that the try-catch block is outside the loop,
so the loop will die on an exception, leaving further messages unprocessed.
* Add ``PostgresSubscribableChannel.errorHandler` option to be invoked
after a `RetryTemplate` and for every failed message.
* The `askForMessage()` new logic is to catch an exception on a message and call `errorHandler`
returning a `FALLBACK_STUB` to continue an outer loop in the `notifyUpdate()`
**Cherry-pick to `6.1.x` & `6.0.x`**
* * Rename private `PostgresSubscribableChannel.askForMessage()` method to more specific `pollAndDispatchMessage()`
The `PostgresSubscribableChannel` uses a task executor
for dispatching messages.
Even if we stop `PostgresChannelMessageTableSubscriber` and
unsubscribe from the channel, the task might be ongoing.
* Use explicit `ThreadPoolTaskExecutor` in the test
to shout it down and wait for tasks to be completed before
verifying DB status
* Optimize `PostgresSubscribableChannel` to mark TX for rollback
when we got a message from DB, but no handlers subscribed
# Conflicts:
# spring-integration-jdbc/src/test/java/org/springframework/integration/jdbc/channel/PostgresChannelMessageTableSubscriberTests.java
* Use `StandardCharsets.UTF_8` as a charset configuration instead of string.
Later remove the necessity of handling encoder errors.
* Use regular expressions to validate the `UUID` string standard representation.
Later obsolete the need for try/catch exceptions.
* Deprecate `UUIDConverter.DEFAULT_CHARSET` since it is out of use.
**Cherry-pick to `6.1.x` & `6.0.x`**
Code clean up for `UUIDConverter`
