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
Fixes https://github.com/spring-projects/spring-integration/issues/8708
According to the `org.apache.sshd.common.channel.ChannelAsyncOutputStream.writeBuffer()` JavaDocs cannot be used concurrently.
* Introduce internal `DefaultSftpSessionFactory.ConcurrentSftpClient` extension
of the `DefaultSftpClient` to set a `Lock` around `super.send(cmd, buffer);`
* Remove lock from the `SftpSession` since it now is managed by the mentioned `ConcurrentSftpClient`
**Cherry-pick to `6.1.x` & `6.0.x`**
# Conflicts:
# spring-integration-sftp/src/main/java/org/springframework/integration/sftp/session/SftpSession.java
In the current code, an `IllegalStateException` might be thrown from the try block while invoking the `removeLockKeyInnerUnlink()` method, especially when caused by key expiration (resulting in `unlinkResult == false`).
This triggers the check block, which incorrectly sets the `unlinkAvailable` flag to `false`, even if the Redis server supports the unlink operation.
As a consequence, the subsequent `removeLockKeyInnerDelete()` method is invoked when it should not be.
* `IllegalStateException` should not be thrown from try block
* Add a comment and fix Checkstyle violations
**Cherry-pick to `6.1.x` & `6.0.x`**
(cherry picked from commit ba6d35d123)
Expected Behavior
Using a single Lua script to verify ownership of the lock and remove it.
Current Behavior
`unlock()` method of `RedisLock` uses two separate Redis operations:
* `isAcquiredInThisProcess()`` method executes a `GET` operation to verify if the lock is owned by the process.
* `removeLockKey()`` method executes `UNLINK/DEL` operation to remove the lock.
* The `removeLockKeyInnerUnlink()`, and `removeLockKeyInnerDelete()`
methods will execute a script both verify ownership of the lock and remove it.
**Cherry-pick to `6.1.x` & `6.0.x`**
# Conflicts:
# spring-integration-redis/src/main/java/org/springframework/integration/redis/util/RedisLockRegistry.java
Fixes https://github.com/spring-projects/spring-integration/issues/8697
The `IntegrationUtils.INTEGRATION_CONVERSION_SERVICE_BEAN_NAME` is not designed
for end-user code autowire, rather internal framework requirement
* Mark `BeanDefinition` for `IntegrationUtils.INTEGRATION_CONVERSION_SERVICE_BEAN_NAME`
as `setAutowireCandidate(false)`
(cherry picked from commit f9dc75c739)
* GH-8685: Re-fetch group after setting condition
Fixes https://github.com/spring-projects/spring-integration/issues/8685
The `AbstractCorrelatingMessageHandler` updates the group metadata in DB
not only for provided `condition`, but also a `lastModified` field.
A subsequent scheduling for group timeout takes the `lastModified`
to compare with the value in the store after re-fetching group in task.
This does not reflect reality since adding `condition` modifies the data in DB,
but in-memory state remains the same.
* Re-fetch a group from the store in the `AbstractCorrelatingMessageHandler.setGroupConditionIfAny()`.
* Verify expected behavior via new `ConfigurableMongoDbMessageGroupStoreTests.groupIsForceReleaseAfterTimeoutWhenGroupConditionIsSet()`
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**
* * Fix Checkstyle violation in the test
