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`
We use an aggregate handler that uses a group timeout.
It's expected that this timeout will be hit frequently.
However, this causes a info log message emitted every time.
I'd propose to log this message on debug.
**Cherry-pick to `6.1.x` & `6.0.x`**
(cherry picked from commit f89ca99bde)
* GH-8745: Add RFT.shouldMarkSessionAsDirty()
Fixes https://github.com/spring-projects/spring-integration/issues/8745
Not all errors caught in the `RemoteFileTemplate.execute()`
are fatal to mark session as dirty and physically close the target session
in the cache
* Introduce a `RemoteFileTemplate.shouldMarkSessionAsDirty()`
to consult with an exception if it is really a fatal error to close
the session in the end.
* Override `shouldMarkSessionAsDirty()` in the `RemoteFileTemplate`
implementations to check statuses of respective protocol errors
**Cherry-pick to `6.1.x` & `6.0.x`**
* * Fix tests for pool interaction
* * Fix language in Javadocs
* Add more `not dirty` statuses to `SftpRemoteFileTemplate` & `SmbRemoteFileTemplate`
Fixesspring-projects/spring-integration#8754
* `AbstractIntegrationMessageBuilder.setExpirationDate(@Nullable Long expirationDate)`
* `MessageBuilder.setExpirationDate(@Nullable Long expirationDate)`
* `MessageBuilder.setExpirationDate(@Nullable Date expirationDate)`
**Cherry-pick to `6.1.x` & `6.0.x`**
(cherry picked from commit 75fcddac00)
Fixes https://github.com/spring-projects/spring-integration/issues/8748
The Oracle DB throws `ORA-08177: can't serialize access for this transaction`
when other transaction on the row has begun
* Change the isolation for `DefaultLockRepository.acquire()` transaction
to the `READ_COMMITTED` for what database automatically and silently
restarts the entire SQL statement, and no error occurs.
* Add `oracle` dependencies to JDBC module
* Introduce `OracleContainerTest` and implement it for `OracleLockRegistryTests`
**Cherry-pick to `6.1.x` & `6.0.x`**
* Fix maxMessagesPerPoll for SourcePollingChAdapter
The `AbstractMethodAnnotationPostProcessor` does not check
for `PollerMetadata.MAX_MESSAGES_UNBOUNDED` before setting
`maxMessagesPerPoll` into a `SourcePollingChannelAdapter`
which in this case must be `1`
Also fix `SourcePollingChannelAdapterFactoryBean` to not mutate
the provided `PollerMetadata` (which might be global default)
with a new `maxMessagesPerPoll`
**Cherry-pick to `6.1.x` & `6.0.x`**
* * Fix `this.` prefix in `SourcePollingChannelAdapterFactoryBean`
It looks like in the latest Hibernate there is some fix
in the persistent context where provided entity is updated
by the values from DB.
* Fix failing tests
* Migrate affected test classes to JUnit 5
**Cherry-pick to `6.1.x`**
# Conflicts:
# build.gradle
It looks like in the latest Hibernate there is some fix
in the persistent context where provided entity is updated by the values from
DB.
We are going to address those failing JPA tests post-release
The `JmsMessageDrivenEndpoint` delegates all the hard work to the
`ChannelPublishingJmsMessageListener`, but missed to propagate an `ObservationRegistry`
and other related options.
The `JmsInboundGateway` is worse: it delegated to the `JmsMessageDrivenEndpoint`
* Add `IntegrationObservation.HANDLER` observation to the `MessagingGatewaySupport.send()`
operation: used by the delegate in the `ChannelPublishingJmsMessageListener`
* Expose and propagate observation-related options from `JmsInboundGateway`
and `JmsMessageDrivenEndpoint`
* Expose `observationConvention()` option on the `MessagingGatewaySpec`
and `MessageProducerSpec`
* Remove unused imports
* Do not start a new `RECEIVER` observation if there is already `SERVER` one
* Fix `MessagingGatewaySupport` for `Observation.NOOP` check.
The parent process may still use `ObservationRegistry.NOOP` which sets
`Observation.NOOP` instance into the current context and thread local.
**Cherry-pick to `6.1.x` & `6.0.x`**
Fixes https://github.com/spring-projects/spring-integration/issues/8713
* Introduce `DefaultSftpSessionFactory.createSftpClient()` factory method
which can be overridden for any custom `SftpClient` use-case
* Add changes to the docs
* Some code clean up
**Cherry-pick to `6.1.x`**
# Conflicts:
# spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/DefaultSftpSessionFactory.java
# src/reference/antora/modules/ROOT/pages/sftp/session-factory.adoc
# src/reference/antora/modules/ROOT/pages/whats-new.adoc
* Fix ThreadSPropagationChInterceptor for stacking
Related SO thread: https://stackoverflow.com/questions/77058188/multiple-threadstatepropagationchannelinterceptors-not-possible
The current `ThreadStatePropagationChannelInterceptor` logic is to wrap one
message to another (`MessageWithThreadState`), essentially stacking contexts.
The `postReceive()` logic is to unwrap a `MessageWithThreadState`,
therefore we deal with the latest pushed context which leads to the `ClassCastException`
* Rework `ThreadStatePropagationChannelInterceptor` logic to reuse existing `MessageWithThreadState`
and add the current context to its `stateQueue`.
Therefore, the `postReceive()` will `poll()` the oldest context which is, essentially,
the one populated by this interceptor before, according to the interceptors order
* Fix `AbstractMessageChannel.setInterceptors()` to not modify provided list of interceptors
* The new `ThreadStatePropagationChannelInterceptorTests` demonstrates the problem
described in that mentioned SO question and verifies that context are propagated
in the order they have been populated
**Cherry-pick to `6.1.x` & `6.0.x`**
* * Fix `ThreadStatePropagationChannelInterceptor` for publish-subscribe scenario.
Essentially, copy the state queue to a new decorated message
* Fix `BroadcastingDispatcher` to always decorate message, even if not `applySequence`
* * Fix unused import in the `BroadcastingDispatcher`
* * Fix unused import in the `ThreadStatePropagationChannelInterceptor`
The `ExpirableLockRegistry.expireUnusedOlderThan()` uses a query to the target store by mistake.
The logic of this API is indeed about the local cache for those lock instances.
We cannot effect any other processes with this cache.
And even if we remove our local instance while it is locked in other process,
that doesn't mean that on the next `obtain()` call, when we got a fresh local instance, we will be able to acquire a lock in target store.
* Fix `isAcquiredInThisProcess()` to check only `localLock.isLocked()`
**Cherry-pick to `6.1.x` & `6.0.x`**
# Conflicts:
# spring-integration-jdbc/src/main/java/org/springframework/integration/jdbc/lock/JdbcLockRegistry.java
When transaction is configured for the `PostgresSubscribableChannel.notifyUpdate()`
and it is rolled back, the next poll in that loop will return the same message.
Again and again if transaction is always rolled back.
This leads to the condition when we never leave this loop even if
we fully unsubscribed from this channel.
The issue has need spotted after introducing `SKIP LOCKED`
for `PostgresChannelMessageStoreQueryProvider` which leads to
the locked record in DB in the mentioned above transaction.
* Introduce `PostgresSubscribableChannel.hasHandlers` flag
to check in the `notifyUpdate()` before performing poll query in DB.
**Cherry-pick to `6.1.x` & `6.0.x`**
* GH-8732 Don't remove JDBC message if other groups
Fixes https://github.com/spring-projects/spring-integration/issues/8732
When same message is added into different groups,
its record in the `INT_MESSAGE` must remain until the last group is removed
* Improve `JdbcMessageStore.DELETE_MESSAGES_FROM_GROUP` SQL
to ignore those messages for removal which has other group records in the `INT_GROUP_TO_MESSAGE` table
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**
* * Fix `JdbcMessageStore.removeMessage()` and `removeMessagesFromGroup()`
to remove from `INT_MESSAGE` only if there is no other records in `INT_GROUP_TO_MESSAGE`
* * Improve `MessageStore.removeMessage()` Javadoc
mentioning `MessageGroupStore` specifics
