Fixes: #9001
The `ObservationPropagationChannelInterceptor` does not propagate an observation properly.
And it fully cannot when the message channel is persistent.
* Deprecate `ObservationPropagationChannelInterceptor` in favor of enabled observation on the channel
and target `MessageHandler` which is a consumer of this channel.
* Remove tests with an `ObservationPropagationChannelInterceptor`
* Mention a correct behavior in the `metrics.adoc` and `ObservationPropagationChannelInterceptor` Javadocs
Fixes: #8950
The `IntegrationFlow.toReactivePublisher(true)` makes `PublisherIntegrationFlow`
to be stopped from the beginning and waiting for the `Publisher.subscribe()`.
However only `EndpointSpec` components are marked as `autoStartup(false)`.
The `MessageProducerSupport` is not included, therefore unexpected messages
are produced to the channel in the end of flow without subscribers.
* Check for `AbstractEndpoint` component type in the `PublisherIntegrationFlow`
and mark it as `setAutoStartup(false)`
* Ensure in a new `ReactiveStreamsTests.messageProducerIsNotStartedAutomatically()` test
that `MessageProducerSupport` is not started from the beginning,
but rather when we subscribe to the `Publisher` via `StepVerifier`
(cherry picked from commit 2b576c57d6)
There is no need to post-process the framework internal BPPs.
Therefore, mark `GatewayProxyInstantiationPostProcessor`, `GlobalChannelInterceptorProcessor`, `IntegrationFlowBeanPostProcessor`
and `BaseIntegrationFlowDefinition.ReplyProducerCleaner` as an `AopInfrastructureBean`.
This way they are not post-processed via AOP `BeanPostProcessor`
* Migrate `org.springframework.integration.aop` tests to JUnit 5
* Fix race condition in the `FluxMessageChannelTests.noMemoryLeakInFluxMessageChannelForVolatilePublishers()`
where `FluxMessageChannel.upstreamSubscriptions` is updated from a different thread
(cherry picked from commit a8ed1dd9aa)
Fixes: #8894
* Guard `this.messageStore` access in the `DelayHandler` with existing `this.lock`
to avoid `ConcurrentModificationException`
**Cherry-pick to `6.2.x`**
(cherry picked from commit 4b43969438)
Fixes: #8889
* Gather the current thread local state for the message into `IntegrationMessageHeaderAccessor.REACTOR_CONTEXT` header
before `this.sink.tryEmitNext(messageToEmit)`
* Restore thread locals for the current message in the `ReactiveStreamsConsumer.SubscriberDecorator` just before `this.delegate.onNext(messageToDeliver)`
**Cherry-pick to `6.2.x`**
(cherry picked from commit dbd46dfe3d)
Related to: #8856
Many tests create their own `ThreadPoolTaskScheduler` beans.
Therefore, its default phase might affect the memory and performance.
* Use `phase = SmartLifecycle.DEFAULT_PHASE / 2` for manual
`ThreadPoolTaskScheduler` beans
* Migrate affected tests classes to JUnit 5
* Make some other configuration adjustments for better performance
**Cherry-pick to `6.2.x`**
(cherry picked from commit 39c99c0719)
Fixes: #8856
The `ThreadPoolTaskScheduler` in Spring Framework manages now a lifecycle
with a `Integer.MAX_VALUE` phase by default.
This causes a wait block for scheduled tasks in the `AbstractPollingEndpoint` instances.
These tasks are cancelled by in the later `Integer.MAX_VALUE / 2` phase.
So, we have a lifecycle deadlock on context close
* Fix `DefaultConfiguringBeanFactoryPostProcessor.registerTaskScheduler()`
with a `.addPropertyValue("phase", SmartLifecycle.DEFAULT_PHASE / 2)`
to let the `AbstractPollingEndpoint` to cancel its currently scheduled task
to avoid possible pollution with unexpected (and lost) messages
**Cherry-pick to `6.2.x`**
(cherry picked from commit 11e0ea1305)
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`**
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`**
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>
* 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`
* 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`**
* 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`
* 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
* GH-2971: Add `LockRegistry.executeLocked()` API
Fixes https://github.com/spring-projects/spring-integration/issues/2971
* Following best practice and well-known patterns with `Jdbc`, `Rest` or `Jms` templates,
introduce `default` methods into `LockRegistry` interface to make it easier to perform
tasks when within a lock.
* Since all the required logic is now covered by those `LockRegistry.executeLocked()` methods,
there is no need in the dedicated abstract `WhileLockedProcessor` class.
Deprecated it for removal in the next version
* Use a new `LockRegistry.executeLocked()` API in the `FileWritingMessageHandler`
instead of just deprecated `WhileLockedProcessor`
* To satisfy Java limitations for checked lambdas, introduce `CheckedCallable` and `CheckedRunnable` utilities
similar to interfaces in the `io.micrometer.observation.Observation`
* Change existing `CheckedFunction` to expose extra generic argument for `Throwable`
* Add dedicated chapter for distributed lock into docs
* Fix some links and typos in the docs
* * Fix Javadoc for `CheckedFunction`
* 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/8705
an internal `MethodInvocationGateway` is a `MessagingGatewaySupport`
extension with all the logic available there.
One of the option introduced in `5.2.2` to be able to throw a `MessageTimeoutException`
instead of returning `null` when no reply received in time from downstream flow
* Expose an `errorOnTimeout` on the `@MessagingGateway` and `GatewayEndpointSpec`
* Propagate this option from a `GatewayProxyFactoryBean` down to its internal
`MethodInvocationGateway` implementation
* Modify couple tests to react for `errorOnTimeout` set to `true`
* Document the feature
Fix language in Docs
Co-authored-by: Gary Russell <grussell@vmware.com>
* GH-8704: Add global property for `defaultTimeout`
Fixes https://github.com/spring-projects/spring-integration/issues/8704
The default timeout for requests and replies in the integration endpoints
is 30 seconds to avoid indefinite blocking in threads.
Sometime those 30 seconds is not enough.
* Introduce a `spring.integration.endpoints.defaultTimeout` global property
to allow overriding all the timeouts to desired value.
The negative number indicates an indefinite waiting time: similar to what
was there before introducing 30 seconds by default
* 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/8703
* Instantiate a `MessagingAnnotationBeanPostProcessor` via factory method from `MessagingAnnotationPostProcessor`
avoiding extra code generation on an explicitly provided complex `Map` for bean definition property
* Fix test to react properly to a new logic of `MessagingAnnotationBeanPostProcessor` bean registration
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)`
* Remove generic argument from the `EnricherSpec.property()` which simply does nothing, but noise
- `Object` is enough
* Do the same for `EnricherSpec.header()`
* Fix `KotlinEnricherSpec`, respectively: no generic argument - just `Any`
* Make `KotlinIntegrationFlowDefinition.route()` to expect non-null for `router` arg
* Add `& Any` to generic params of the `KotlinRouterSpec` methods - according compiler warning requirements
* 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
Fixes https://github.com/spring-projects/spring-integration/issues/8664
When an `Observation` is turned to `NoopObservation`,
the `observeWithContext()` fails with `ClassCastException`
since `NoopObservation` serves just plain `Context` not the one we supplied
* Fix `MessagingGatewaySupport.sendAndReceiveWithObservation()` same way
as it is in version `6.0.x`
* Modify `IntegrationObservabilityZipkinTests` to reject some `Observation`
via `observationPredicate()` configuration
**Cherry-pick to `6.1.x`**
* Introduce a `SplitterSpec` which can accept possible
splitter variants: `expression`, `function`, `ref` etc.
This way we are going to have a complex endpoint configuration
only with a single method argument.
* Use `SplitterSpec` in a newly introduced `splitWith()`
* Deprecate those `split()` methods which use `SplitterEndpointSpec`.
This method are complex enough because of their several arguments
* Refactor some common `MessageHandler` options initialization
into the `ConsumerEndpointSpec.doGet()`
* Disable failing now `MethodInvokingMessageProcessorTests.testCollectionArgument()`:
or Jackson problem, or fresh SF
* Groovy DSL will be fixed in the separate PR:
https://stackoverflow.com/questions/76595843/groovy-selects-a-defaultgroovymethods-split-instead-of-mine-one
Fixes https://github.com/spring-projects/spring-integration/issues/8611
According to the latest Spring Framework requirements related to AOT the
`BeanDefinitionRegistryPostProcessor` must not do anything but only bean registrations
* Extract a `MessagingAnnotationBeanPostProcessor` component with a `BeanPostProcessor` logic
from the `MessagingAnnotationPostProcessor` and leave the last one as just
a `BeanDefinitionRegistryPostProcessor` impl
* Fix tests according to a new logic
Fixes https://github.com/spring-projects/spring-integration/issues/8611
According to the latest Spring Framework requirements related to AOT the
`BeanDefinitionRegistryPostProcessor` must not do anything but only bean registrations
* Extract an `IntegrationConfigurationReport` component
and move `IntegrationProperties` printing over here out from the
`DefaultConfiguringBeanFactoryPostProcessor`
* Remove a `SmartInitializingSingleton` impl from the `DefaultConfiguringBeanFactoryPostProcessor`
The `ConsumerEndpointSpec` extensions for Kotlin
don't delegate to the provided `endpointFactoryBean`
* Introduce `KotlinConsumerEndpointSpec` extension for `ConsumerEndpointSpec`
with the proper delegation to the provided spec
* Use `KotlinConsumerEndpointSpec` in the Kotlin-specific `Spec` classes
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**
Fixes https://github.com/spring-projects/spring-integration/issues/8654
Spring Framework has an ability to start dependant beans automatically
when we start the current one.
The `AbstractMethodAnnotationPostProcessor` is missing a bean dependency
registration causing errors in target applications when Messaging Annotations configuration is used.
* Add `registerDependentBean()` into an `AbstractMethodAnnotationPostProcessor` when we generate
and register a `ConsumerEndpointFactoryBean`
* Change one of the `ClientManagerBackToBackTests` configuration to rely on a `@ServiceActivator`
for `Mqttv5PahoMessageHandler` bean to ensure that change in the `AbstractMethodAnnotationPostProcessor`
has a proper effect
**Cherry-pick to `6.1.x` & `6.0.x`**
* GH-8626: Provide cleaner `transform()` DSL
Fixes https://github.com/spring-projects/spring-integration/issues/8626
* Add missed `transform(String beanName, @Nullable String methodName)` API
* Introduce a `TransformerSpec` to expose a strict API to configure
transformer variants.
* Introduce `transformWith(Consumer<TransformerSpec>)` as a single point
of all possible transformer and its endpoint options
* Deprecate those `IntegrationFlowDefinition.transform()` variants
which are harder to configure as several lambda arguments
This change will make Kotlin & Groovy DSL more readable and straightforward
* * Use new `transformWith()` in tests where a deprecated API still used
* Add JavaDocs to `TransformerSpec`
* Fix generic types for `BaseIntegrationFlowDefinition.transformWith()` -
they make sense exactly on the `TransformerSpec.transformer()` only
* Apply `transformWith()` for Groovy DSL
* Introduce a new `ClassUtils.isLambda(Object candidate)`
and add a check for Groovy `Closure`
* Fix `GroovyIntegrationFlowDefinition.createConfigurerIfAny()` to propagate a `Consumer` argument
down to the `Closure`
* * Rename `TransformerSpec -> TransformerEndpointSpec` for better context
meaning of the class
* Introduce `KotlinTransformerEndpointSpec` as an extension of the `TransformerEndpointSpec`
to have an `inline fun <reified P> transformer(crossinline function: (P) -> Any)`
for Kotlin style
* Add `KotlinIntegrationFlowDefinition.transformWith(KotlinTransformerEndpointSpec)`
* Deprecate Kotlin methods which are covered by the mentioned `transformWith()`
* Fix tests to use new API
* Mentioned the change in the doc
Related to https://github.com/spring-projects/spring-integration/issues/8644
With virtual threads it is recommended to stay away from `ThreadLocal`
variables to avoid memory exhausting with too many virtual threads
* Fix `RequestHandlerRetryAdvice` to transfer the message context via internal
`IntegrationRetryCallback` implementation.
* Cast to this `IntegrationRetryCallback` in a newly introduced internal `IntegrationRetryListener`
to extract `messageToTry` and set it into a `RetryContext`
`ErrorMessageUtils.FAILED_MESSAGE_CONTEXT_KEY` attribute
* Deprecate a usage of an external `RetryListener` implementation of the `RequestHandlerRetryAdvice`
The `NumberFormatException` flow control is costly
operation
* Use `Character.isDigit()` check iterating through property String
instead of `NumberFormatException` flow control
**Cherry-pick to `6.1.x`, `6.0.x` & `5.5.x`**
* GH-8642: Revise executors in the project
Fixes https://github.com/spring-projects/spring-integration/issues/8642
* Rework some `Executors.newSingleThreadExecutor()` to `ExecutorServiceAdapter(new SimpleAsyncTaskExecutor())`
* Expose `TaskExecutor` setters; deprecate `ExecutorService`-based
* Some other code clean up in the effected classes: `LogAccessor`, no `synchronized` in critical blocks
* Give a meaningful prefix for default threads in the context of components, e.g. `SubscribableRedisChannel` - `getBeanName() + "-"`
* * Fix `PostgresChannelMessageTableSubscriberTests` for
`PostgresSubscribableChannel` initialization to let it create
its internal `Executor`
* Use an `AsyncTaskExecutor` injection instead of `ExecutorServiceAdapter` wrapping
* Fix `LockRegistryLeaderInitiatorTests` for `taskExecutor` injection
* Bring back `LockRegistryLeaderInitiator.setExecutorService()`
as an accident after property auto-renaming
* Mention in the Javadocs of the `DelayHandler`, `DelayerEndpointSpec`,
`BaseIntegrationFlowDefinition.delay()`, `GroovyIntegrationFlowDefinition.delay()`,
`KotlinIntegrationFlowDefinition.delay()` that `messageGroupId` is required option
* Explain in the docs why `messageGroupId` is required and why it cannot rely on a bean name
* Improve Delayer DSL
Move `groupId` option from a `delay()` method arg to the `DelayerEndpointSpec`
to make it cleaner from code reading perspective
* Expose new DSL method based on just a `DelayerEndpointSpec` for Kotlin &v Groovy
* Deprecate multi-arg `delay()` methods in favor of `Consumer<DelayerEndpointSpec>`-based
* * Fix language and code style
Even if the `PartitionedDispatcher.populatedPartitions()`
is fast, in-memory, non-blocking operation, its active call from the `dispatch()`
on every message sent to the channel may pin the virtual thread.
* Optimize the `populatedPartitions()` for double `if`
where we will step into a `synchronized` block only for first several concurrent messages
**Cherry-pick to `6.1.x`**
The `IntegrationFlowBeanPostProcessor.processIntegrationComponentSpec()`
uses a wrong `generateBeanName()` for component to register making the
provided `id` as a prefix
* Use `generateBeanName(Object instance, String prefix, @Nullable String fallbackId, boolean useFlowIdAsPrefix)`
instead to properly "fallback" to the provided name
**Cherry-pick to `6.1.x`**
