Fixes: #9866
Issue link: https://github.com/spring-projects/spring-integration/issues/9866
Due to race condition between `sink.tryEmitNext()` and `sink.emitComplete()``,
there could be a situation when `onNext` signal slips after `onComplete`.
Appears on fast producers into this `FluxMessageChannel`.
That leads to error like `Spec. Rule 1.3 - onSubscribe, onNext, onError and onComplete signaled to a Subscriber MUST be signaled serially.`
* Fix `FluxMessageChannel` to check for `this.active` one more time just before `sink.tryEmitNext()` call
* Also mitigate a race condition with a `Sinks.EmitFailureHandler.busyLooping(Duration.ofSeconds(1))` in the `destroy()`
instead of `Sinks.EmitFailureHandler.FAIL_FAST`.
The `busyLooping()` would retry requested `onComplete()` signal for that specific error until success or timeout
(cherry picked from commit 7680e02cbb)
Fixes: #9863
Issue link: https://github.com/spring-projects/spring-integration/issues/9863
The `PollerMetadata.sendTimeout` has been migrated to the `SourcePollingChannelAdapter.setSendTimeout()` long time ago.
Right now this option is not propagated anywhere
* Introduce missed `SourcePollingChannelAdapterSpec.sendTimeout`
* Deprecate (for removal) `PollerMetadata.sendTimeout`
* Verify `SourcePollingChannelAdapterSpec.sendTimeout` option in the `ReactiveStreamsTests`
(cherry picked from commit 0af3b25cd6)
# Conflicts:
# spring-integration-core/src/main/java/org/springframework/integration/dsl/SourcePollingChannelAdapterSpec.java
Fixes: #9825
Issue link: https://github.com/spring-projects/spring-integration/issues/9825
The `DelayerEndpointSpec` extends `ConsumerEndpointSpec` which has a `taskScheduler()` option.
However this is set only to the endpoint for this `MessageHandler`.
* Override `taskScheduler()` method on the `DelayerEndpointSpec` to set
the provided `TaskScheduler` to the `DelayHandler` as well
(cherry picked from commit 12fee0a9fb)
Fixes: #9745
Issue link: https://github.com/spring-projects/spring-integration/issues/9745
The `IntegrationEvaluationContextFactoryBean` does not provide a `TypeLocator` by default.
That may lead to a class-not-found problem from different `ClassLoader` in the parallel Java `Stream` executor.
* Fix `IntegrationEvaluationContextFactoryBean` to use a `StandardTypeLocator` based on the `applicationContext.getClassLoader()`
(cherry picked from commit 6184c4082a)
# Conflicts:
# spring-integration-core/src/main/java/org/springframework/integration/config/IntegrationEvaluationContextFactoryBean.java
Fixes: #9709
Issue link: https://github.com/spring-projects/spring-integration/issues/9709
The Java DSL loses an `inputChannel` when existing channel is referenced by its name.
Then the target IntegrationFlow does not contain a bean reference for this channel and when we call its `getInputChannel()` we got something from middle of the flow.
However, that `inputChannel` is really expected to be an input for the flow.
* Fix `IntegrationFlowBeanPostProcessor` to get a bean by `MessageChannelReference` and populate it into `targetIntegrationComponents`
as we do for newly created `DirectChannel` if there is no bean for provided `MessageChannelReference`
# Conflicts:
# spring-integration-core/src/main/java/org/springframework/integration/dsl/context/IntegrationFlowBeanPostProcessor.java
Fixes: #9705
Issue link: https://github.com/spring-projects/spring-integration/issues/9705
Currently when any class that implements `AbstractReplyProducingMessageHandler` doesn't produce a reply it will log the message, even if no reply is required.
The message should only be logged if `isLoggingEnabled()` returns true as a handler that doesn't require a reply may be a normal operation that is expected, and if we've set `loggingEnabled` to false it shouldn't log the message.
* Prevent logging when no reply is provided for an `AbstractReplyProducingMessageHandler` if logging is disabled
(cherry picked from commit 7fc104ac8b)
Fixes: #9702
Issue link: https://github.com/spring-projects/spring-integration/issues/9702
The current `CheckedCallable.unchecked()` returns `Runnable`, which is not an expectation.
* Deprecate `CheckedCallable.unchecked()` in favor of newly introduced `CheckedCallable.uncheckedCallable()`.
We cannot call it `unchecked()` as well, since `Callable` after erasure becomes similar to class signature as `Runnable`.
(cherry picked from commit 8c22bf6430)
Additional code style cleanups in the affected classes
**Auto-cherry-pick to `6.2.x`**
# Conflicts:
# spring-integration-core/src/test/java/org/springframework/integration/json/JsonPathTests.java
Such a `Thread.sleep()` makes tests slower, plus may cause a longer timing issue on slow CI/CD when CPU resources are limited
(cherry picked from commit 09750f2012)
Fixes: #9623
Issue link: https://github.com/spring-projects/spring-integration/issues/9623
The `ConcurrentModificationException` is thrown from the `ThreadStatePropagationChannelInterceptor.MessageWithThreadState.stateQueue`
which is a not thread-safe `LinkedList`
* Fix `ThreadStatePropagationChannelInterceptor.MessageWithThreadState.stateQueue` to be a `LinkedBlockingQueue` instead
(cherry picked from commit ba57ee8a1b)
Fixes: #9614
Issue link: https://github.com/spring-projects/spring-integration/issues/9614
The `Matcher.quoteReplacement()` escapes `\` as well as `$`.
However, Jackson tries to resolve special symbol from the escaped `$` and fails as `Unrecognized character escape '$' (code 36)`
* Fix `SimpleJsonSerializer.toElement()` to escape only `\`
(cherry picked from commit 8dad15bac2)
Fixes: #9561
PR: https://github.com/spring-projects/spring-integration/pull/9561
The `DelayHandler.DelayedMessageWrapper` cannot be deserialized when using `RedisMessageStore` and JSON serialization:
```
org.springframework.data.redis.serializer.SerializationException: Could not read JSON:Cannot construct instance of `org.springframework.integration.handler.DelayHandler$DelayedMessageWrapper` (no Creators, like default constructor, exist): cannot deserialize from Object value (no delegate- or property-based Creator)
at [Source: UNKNOWN; byte offset: #UNKNOWN]
at org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer.deserialize(GenericJackson2JsonRedisSerializer.java:311)
```
* More code clean up and refactoring in the test
(cherry picked from commit 73bb813b53)
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`**
