Fixes: #9478
Issue link: https://github.com/spring-projects/spring-integration/issues/9478
Any `MessagingGatewaySupport` implementation does not register itself into a `SmartLifecycleRoleController`
because they don't call `super.onInit()` of the `AbstractEndpoint`
* Fix `MessagingGatewaySupport` for calling `super.onInit()` from its `onInit()`
* Verify `SmartLifecycleRoleController` registration in the `KafkaInboundGatewayTests`
* Remove out of use XML `group-id` attribute from Kafka channel adapter
and move it to the `channel` XSD as the place where it is really used
(cherry picked from commit a8174d5bce)
Even if `Executors.newSingleThreadExecutor()` returns a `FinalizableDelegatedExecutorService`,
an instance is kept in the memory until JVM exists.
That may lead to memory leak since we have a lot of threads in memory.
(cherry picked from commit fdac8f1634)
* Add `@DirtiesContext` to the `RecipientListRouterParserTests`
* Remove manual `context.start();` since it is started by the `@SpringJUnitConfig` anyway
(cherry picked from commit 777c84f603)
Looks like `ApplicaitonContext.stop()` is note called on `close()`.
This might cause an `OutOfMemoryError` on the CI
* Use Spring context configuration for the test environment
(cherry picked from commit b2a3d25ffd)
Fixes: #9335
Spring Boot provides a `CompositeMeterRegistry` bean marked with a `@Primary`.
So, this one has to be autowired whenever we would like to deal with application
environment.
The `MicrometerNodeEnhancer` has a flaw to skip `MeterRegistry` injection if we have more than one `MeterRegistry` bean.
* Fix `MicrometerNodeEnhancer` logic to use `ObjectProvider.getIfUnique()` API
which is able to deal with the `primary` properly.
Otherwise, `null` as it was before
(cherry picked from commit e0472d887a)
Fixes: #9259
The `Mono.toFuture()` does not propagate context to thread locals of the `CompletableFuture` consumer.
See `MonoToCompletableFuture`
* Fix `AbstractMessageProducingHandler` to convert reply `Mono` to `CompletableFuture` manually.
Use `doOnEach()` and set thread locals from the Reactor context manually around `replyFuture.complete()/completeExceptionally()`
* Add respective unit test into `WebFluxObservationPropagationTests` to ensure that same trace is used in downstream endpoints after WebFlux client reply
(cherry picked from commit 4a77dbcc46)
Fixes: #9260
The current `IntegrationManagementConfiguration.obtainObservationPatterns()` is to always have an `*` as a first
pattern in the final result.
Because of `HashSet` natural order.
This would lead to ignore all the patterns, especially those with negation
* Fix `IntegrationManagementConfiguration.obtainObservationPatterns()` logic to ignore regular
patterns if `*` is present.
Optimization wise.
* Always put `*` into the end of final array let negative patterns to do their job
* Modify `IntegrationGraphServerTests` & `WebFluxObservationPropagationTests`
test configurations to ensure that `*` pattern does not have a precedence over negative patterns.
(cherry picked from commit 4108ea520e)
The `Sinks.Many<Boolean> subscribedSignal` is a bit of overhead in the logic.
* Use `Mono.fromCallable(this.sink::currentSubscriberCount)` instead for `delaySubscription()`
when we perform `subscribeTo(Publisher)`
(cherry picked from commit bdcb856a90)
Fixes: #9191
Not all inbound protocol handlers convert header from native representation.
Sometimes they just come in as `byte[]`.
* Fix `MessageReceiverContext` to support `byte[]` for headers used in the observation,
e.g. `B3-*`, `traceparent` etc.
(cherry picked from commit b29ff9a710)
Fixes: #9215
* Instead of `share()` use `publish(1).refCount()` to prefetch only item from upstream.
* Also remove `publishOn(this.scheduler)` for upstream publishers in favor of opt-in on the consumer side.
(cherry picked from commit e9561b41f1)
Fixes: #9198
When observation is enabled on the `MessageChannel`, the message to send is converted to a `MutableMessage`.
In case of `ErrorMessage` this causes a loss of `originalMessage` and may lead to `ClassCastException`
in the target error handler.
* Check for `ErrorMessage` in the `AbstractMessageChannel.sendWithObservation()` and create a new one
as a copy of original request, but including observation headers before performing `sendInternal()`
* Modify `IntegrationObservabilityZipkinTests` to verify an observation with an `ErrorMessage` and its handler.
(cherry picked from commit b0cbacfafb)
Related to: #9110
* Make `IntegrationReactiveUtils.CONTEXT_SNAPSHOT_FACTORY` instance conditional on the `IntegrationReactiveUtils.isContextPropagationPresent`
to avoid class-not-found problem for the `io.micrometer.context.ContextSnapshot`
* Make `IntegrationReactiveUtils.captureReactorContext()` & `IntegrationReactiveUtils.setThreadLocalsFromReactorContext()`
conditional on the `isContextPropagationPresent`
**Auto-cherry-pick to `6.2.x`**
Fixes: #9110
* Move context propagation utilities to the `IntegrationReactiveUtils`
* Capture context into message header in the `IntegrationReactiveUtils.adaptSubscribableChannelToPublisher()`
before `sink.tryEmitNext()`
* Restore the context from message header in the `flatMap()` for `ReactiveStreamsConsumer.reactiveMessageHandler`
**Auto-cherry-pick to `6.2.x`**
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
**Auto-cherry-pick to `6.2.x`**
Fixes: #8981
Sometime the simple `boolean failover` on the `MessageChannel` (default `true`)
is not enough to be sure that we can dispatch to the next handler or not.
Such a decision can be made using `ErrorMessageExceptionTypeRouter`, but that would
require an overhaul for the whole integration flow
* Introduce a simple `Predicate<Exception> failoverStrategy` into `UnicastingDispatcher`
and all its `MessageChannel` implementation consumers to allow to make a decision about next failover
according to a thrown exception from the current `MessageHandler`
* Expose `failoverStrategy` on the `DirectChannel`, `ExecutorChannel` & `PartitionedChannel`,
and add it into respective Java DSL specs
* Fix involved tests to rely on the `failoverStrategy` property from now on
* Document the new feature
Fixes: #7925
The `MessageHistory.write()` creates not only a new instance of the `MessageHistory`,
but also a new copy of the whole message.
This significantly impacts the performance when we have too many components to track
* Make `MessageHistory` as append-only container and create a new instance (plus message)
only on the first track when no prior history is present
* Change `WireTap`, `BroadcastingDispatcher`, `AbstractMessageRouter` and `AbstractMessageSplitter`
to use a new `AbstractIntegrationMessageBuilder.cloneMessageHistoryIfAny()` API for every branch a message
is produced.
Essentially, create a new message with copy of the message history to let that downstream sub-flow
have its own trace
* Modify failed unit tests for a new logic where message history header is not immutable anymore
* This also fixes an `AbstractMessageSplitter` for propagating its track into messages it emits
* * Do not clone message history header if only one consume in multi-publish
* Fix typos in docs
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`
**Auto-cherry-pick to `6.2.x` & `6.1.x`**
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
**Auto-cherry-pick to `6.2.x`**
Fixes: #8894
* Guard `this.messageStore` access in the `DelayHandler` with existing `this.lock`
to avoid `ConcurrentModificationException`
**Cherry-pick to `6.2.x`**
Fixes: #8893
Simplify the target flows logic for `ExpressionEvaluatingRequestHandlerAdvice.failureChannel`
when no exception is re-thrown and target flow would like to produce a reply.
Currently, there is need in extra logic like `.enrichHeaders(e -> e.replyChannelExpression("payload.failedMessage.headers[replyChannel]"))`
* Copy failed `Message` headers into an `ErrorMessage` before sending it into the `failureChannel` from the `ExpressionEvaluatingRequestHandlerAdvice`
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`**
The `ReactiveInboundChannelAdapterTests.testTimeSupplierConsistency()` subscribes a bit later than `@InboundChannelAdapter`
might have started to emit messages.
* Add `autoStartup = "false"` to the `timeEndpoint` and start in the test only when `StepVerifier` is subscribed
* Add `stop()` for both endpoints in the test configuration to avoid memory pollution with not processed messages in between tests
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`**
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`**
* Remove `forkEvery = 1000` since where the build fails there is no way find the reason
* Rework affected test classes to use JUnit 5 API
* Use Java text block for JSON snippets in the test
* Upgrade supported dependencies to their latest versions or `alpha`, `RC`
* Migrate to `com.github.spotbugs` plugin
* Upgrade to GraalVM `23.1.1`, which includes `org.graalvm.polyglot:js` migration
* Remove `spring-integration-security` module
* Fix Debezium tests according to the latest changes in the Debezium library
* Fix deprecations and incompatibility for latest Smack and Spring AMQP
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`
