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`**
* KafkaMessageDrivenChannelAdapter's ATTRIBUTES_HOLDER should be isolated.
In order to achieve custom retry in batch mode, we may to use a RetryTemplate in listener itself. But if the RetryTemplate is shared with another KafkaMessageDrivenChannelAdapter, batch mode's ATTRIBUTES_HOLDER might be over-written by another KafkaMessageDrivenChannelAdapter's IntegrationRecordMessageListener.
The situation is like shown below.
- There is only one RetryTemple bean in the application.
- There are two KafkaMessageDrivenChannelAdapters(A,B) in the application.
- A KafkaMessageDrivenChannelAdapter is batch mode and utilizing the retryTemplate in the listener.
- B KafkaMessageDrivenChannelAdapter is record mode and using the retryTemplate itself.
- (B KafkaMessageDrivenChannelAdapter's recordListener is registered in the retryTemplate.)
- When A retry is attempted in the listener, it will trigger B KafkaMessageDrivenChannelAdapter's recordListener.
- B KafkaMessageDrivenChannelAdapter's recordListener will overwrite A KafkaMessageDrivenChannelAdapter's ATTRIBUTES_HOLDER.
* should not mutate an externally provided RetryTemplate
* should not expose KafkaInboundEndpoint's methods outside the class and fix a checkstyle error.
* removing unused code and polishing
* restore retry around batch and bring back KafkaInboundEndpoint to endpoints.
* remove retry logic in batch mode and move ATTRIBUTES_HOLDER into KafkaInboundEndpoint
* remove generic type parameters of KafkaInboundEndpoint
* fix style error
---------
Co-authored-by: kurt <kurt@weversecompany.com>
* 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
The `SecurityContextPropagationChannelInterceptor` has been migrated to `spring-security-messaging`.
Since it was only the class in the `spring-integration-security`, it is now fully considered
as deprecated
* Remove all the tests from `spring-integration-security`
* Modify `HttpDslTests` to demonstrate the `spring-security-messaging` in action
which has been replaced with whatever there was in `spring-integration-security`
* Remove redundant `exclude group: 'org.springframework'` for security
dependencies in `build.gradle` since all of them rely on the same SF deps
as SI
Fixes https://github.com/spring-projects/spring-integration/issues/8760
* Make `PostgresChannelMessageStoreQueryProvider` to use single `DELETE ... RETURNING` for polling statements
* Add `isUsingSingleStatementForPoll` and use it from `JdbcChannelMessageStore`
* Execute Postgres init scripts to `PostgresContainerTest`
* Code clean up
* Document the new feature
Fixes https://github.com/spring-projects/spring-integration/issues/8757
* Add ctor `ObjectStringMessageConverter(Charset)`
* create tests for `ObjectStringMessageConverter`
* fix codestyle issues
* revise JavaDoc
* `@NonNullApi` for `org.springframework.integration.support.converter`
* revise `SimpleMessageConverter`
* `inboundMessageMapper` and `outboundMessageMapper` are initialized to their default implementations
Bboth are set in non-default ctors with final setters
* remove `@Nullable` from `private BeanFactory beanFactory`
* Handle empty `String... headerNames`
* Some code clean up
* 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`**
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`**
Related to: https://stackoverflow.com/questions/77268009/how-to-use-sftp-spring-integration-mput-with-sftpoutboundgateway-when-file-objec
In some scenarios when the flow starts with a file inbound channel adapter
and then an MPUT operation is performed for remote file outbound gateway,
the populated in the beginning `FileHeaders.FILENAME` is used from
the `DefaultFileNameGenerator` for all the files from local directory to
upload.
Such a behaviour leads only to the last file in the target remote
directory and only with the name from that header.
* Fix the `AbstractRemoteFileOutboundGateway` to remove a `FileHeaders.FILENAME` header
when message is build for specific item from MPUT request.
This way an original local file is used when we upload directory.
* 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`
Turns out the `providers.provider` does not cache its value
and `grgit.status()` is called for every `updateCopyrights`
task in sub-modules
* Bring back regular `modifiedFiles` property, but guard
it with a `!isCI` condition to avoid the `grgit.status()`
call on CI
* Fix `AbstractAmqpChannel` to add messaging packages
as allowed for deserialization
* Some code style clean up in the `AbstractSubscribableAmqpChannel`
* Disable Kraft (default) for Kafka tests since they
are not reliable(perhaps only on Windows)
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`**
* Change the default polling trigger to 1 second
The current default trigger for the poller is 10 milliseconds fixed delay.
This is very tight policy for Microservices where we might not have too many
scheduled threads to distribute polling endpoint jobs evenly.
* Change the default trigger to 1 second to align with what Spring Boot already
claims.
Same 1 second policy is used in Spring Cloud Stream as well
* Fix language in Docs
Co-authored-by: Gary Russell <grussell@vmware.com>
---------
Co-authored-by: Gary Russell <grussell@vmware.com>
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`
* Fix Checkstyle violations for `private` inner classes where they have to be also `final`
* Fix deprecations after upgrades
* Remove manual coroutines handling in the `GatewayProxyFactoryBean`
in favor of the same functionality in the `JdkDynamicProxy`
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`**
* Fix KV Stores for same message in multiple groups
If same message is stored into different groups with the same KV store,
the removal of one group would lead to removal the message for the other one.
* Improve KV Store to save message with the key including a group id
* Respectively, refine the removal API to include group id into keys
* Also change the `MESSAGE_GROUP_KEY_PREFIX` for group records to `GROUP_OF_MESSAGES_`
since the `MESSAGE_` prefix includes group records as well for various operations
based on key pattern
* * Use `[^GROUP_]` in the pattern for messages count
**Cherry-pick to `6.1.x`**
* * Fix MongoDB MS for message removal logic
* * Bring back `GROUP_OF_MESSAGES_` prefix
to avoid complex regexp and don't bother for edge cases,
where even that regexp may fail
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`**
* 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`**
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/8691
* Remove setAge with TimeUnit
Turned out we don't need anymore since we're using Duration for age in FtpLastModifiedFileListFilter and SftpLastModifiedFileListFilter.
* Add changes to the docs
* Introduce AbstractLastModifiedFileListFilter
* Some code readability improvements
* Make language in the docs more official
