Commit Graph

2921 Commits

Author SHA1 Message Date
Mitchell
02b58277f4 GH-9369: Fix MutableMessageHeaders for serialization
Fixes: #9369 

* Implement `MutableMessageHeaders.readResolve()` to reinstate the instance

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
2024-08-08 17:16:49 +00:00
Sanghyuk Jung
32f2668260 Remove duplicated "the" in Javadocs 2024-08-02 19:26:09 -04:00
Artem Bilan
6054e8bc9f More test clean up to minimize memory
Probably `@DirtiesContext` would remove application contexts for the JUnit cache
2024-07-24 16:45:48 -04:00
Artem Bilan
777c84f603 An attempt to fix OOM in the RecipientListRouterParserTests
* Add `@DirtiesContext` to the `RecipientListRouterParserTests`
* Remove manual `context.start();` since it is started by the `@SpringJUnitConfig` anyway
2024-07-16 11:59:35 -04:00
Artem Bilan
b2a3d25ffd Migrate ReturnAddressTests to JUnit 5
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
2024-07-16 11:55:50 -04:00
Artem Bilan
e0472d887a GH-9335: Fix MicrometerNodeEnhancer for a primary MeterRegistry bean
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

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
2024-07-16 11:16:06 -04:00
Artem Bilan
0461167aba Fix concurrency for StandardIntegrationFlowContext
If we don't get access to `BeanFactory` in exclusive mode,
the `getBean(Class<?>)` may fail with a `BeanCurrentlyInCreationException`
when we register several `IntegrationFlow` instances in parallel
2024-07-12 12:41:13 -04:00
Artem Bilan
1fe40cf2fe Fix failing tests according new changes in SF SpEL module
Related to: https://github.com/spring-projects/spring-framework/issues/33174
2024-07-10 17:43:48 -04:00
Artem Bilan
4e7caed979 Add comment why and how we handle Mono->CompletableFuture 2024-07-03 12:42:10 -04:00
Artem Bilan
4a77dbcc46 GH-9259: Fix Reactor context propagation on reactive reply (#9284)
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

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
2024-06-28 14:01:46 -04:00
Artem Bilan
4108ea520e GH-9260: Fix IntegrationManagementConfiguration.obtainObservationPatterns() logic
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.

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
2024-06-25 16:25:39 -04:00
Artem Bilan
bdcb856a90 Revise subscribed state in the FluxMessageChannel
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)`

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
2024-06-25 14:47:12 -04:00
Artem Bilan
a5c449580b Rethrow an exception in the ManualFlowTests.testConcurrentRegistration()
The test fails sometimes, but its exception in the executor is lost.
Rethrow `exceptionHappened` to be reported into logs
2024-06-24 12:13:03 -04:00
Artem Bilan
b29ff9a710 GH-9191: Support byte[] for headers in the MessageReceiverContext
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.

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
2024-06-07 09:59:49 -04:00
Artem Bilan
e9561b41f1 GH-9215: Honor back-pressure in FluxMessageChannel
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.

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
2024-06-06 11:38:45 -04:00
Artem Bilan
66ca557a54 Remove extra empty line in the IntegrationObservabilityZipkinTests
**Auto-cherry-pick to `6.3.x`**
2024-06-03 15:25:29 -04:00
Artem Bilan
b0cbacfafb GH-9198: Fix MessageChannel observation for ErrorMessage
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.

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
2024-06-03 14:49:33 -04:00
Artem Bilan
8b2da4f996 Start version 6.4
* Move `whats-new` to `changes-6.2-6.3.adoc`
* Upgrade dependency to the latest (including Gradle)
* Move Spring dependencies to based on SF-6.2
* Fix some deprecations and incompatibility
2024-05-28 20:07:35 -04:00
Artem Bilan
3a49b84559 Fix IntegrationReactiveUtils Javadoc
**Auto-cherry-pick to `6.2.x`**
2024-05-06 09:34:13 -04:00
Artem Bilan
2225b83957 Fix IntegrationReactiveUtils for classpath logic
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`**
2024-05-06 09:29:18 -04:00
Artem Bilan
7abf1a53a7 GH-9110: propagate Reactor context to ReactiveMessageHandler
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`**
2024-05-03 16:57:21 -04:00
Artem Bilan
0613f0c092 Fix Checkstyle for MessageProducerSpec.role(String)
**Auto-cherry-pick to `6.2.x` & `6.1.x`**
2024-04-08 16:28:14 -04:00
Artem Bilan
fd81adff27 Expose MessageProducerSpec.role(String) option
Related to: https://stackoverflow.com/questions/78294542/role-in-integrationflow-dsl

**Auto-cherry-pick to `6.2.x` & `6.1.x`**
2024-04-08 16:17:30 -04:00
Artem Bilan
c155d5d418 Add EmptyLineSeparator Checkstyle rule
* Support "blank lines around" via `spring-framework.xml` IDEA config
* Fix all the Checkstyle violations
2024-03-27 16:54:25 -04:00
Artem Bilan
69332a135b Upgrade some deps to their major/minor versions
* Manage `com.google.protobuf` to generate classes from Protos on the fly
* Fix Jackson API usage after its upgrade
2024-03-18 17:52:42 -04:00
Artem Bilan
074d544e94 GH-9001: Revise Observation propagation over the channel
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`**
2024-03-18 16:19:00 -04:00
Artem Bilan
ffe3605c90 GH-8981: Add UnicastingDispatcher.failoverStrategy option
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
2024-03-12 18:16:02 -04:00
Artem Bilan
3849d6adde Fix Kotlin DSL inconsistency & ambiguity 2024-03-12 12:13:25 -04:00
Artem Bilan
2731e9411e GH-7925: Make message history header as mutable
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
2024-03-05 15:05:34 -05:00
Artem Bilan
2b576c57d6 GH-8950: Fix PublisherIntegrationFlow for AbstractEndpoint
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`**
2024-02-23 13:01:38 -05:00
Artem Bilan
a8ed1dd9aa Optimize BPPs as AopInfrastructureBean
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`**
2024-02-09 17:35:06 -05:00
Artem Bilan
4b43969438 GH-8894: Fix race condition in the DelayHandler
Fixes: #8894

* Guard `this.messageStore` access in the `DelayHandler` with existing `this.lock`
to avoid `ConcurrentModificationException`

**Cherry-pick to `6.2.x`**
2024-02-07 16:41:10 -05:00
Artem Bilan
34bdff04c4 GH-8893: ExpERHA: Copy headers into ErrorMessage
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`
2024-02-07 14:18:27 -05:00
Artem Bilan
dbd46dfe3d GH-8889: FluxMCh: propagate context per message
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`**
2024-02-07 11:19:53 -05:00
Artem Bilan
7b2d8a4ba7 Fix race condition in the ReactiveInboundChannelAdapterTests
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
2024-02-02 09:14:14 -05:00
Artem Bilan
0ba60bc752 Close TaskScheduler in tests
* Add `@DirtiesContext` to some tests where `TaskScheduler` is used
2024-01-30 14:29:40 -05:00
Artem Bilan
ae5472e2f5 GH-8863: Remove deprecated API from previous versions
Fixes: #8863
2024-01-17 12:31:43 -05:00
Artem Bilan
39c99c0719 Fix phase for TaskScheduler instances in 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`**
2024-01-16 11:26:51 -05:00
Artem Bilan
11e0ea1305 GH-8856: Adjust phase for TaskScheduler global bean
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`**
2024-01-12 13:51:45 -05:00
Artem Bilan
40d4150dae Fix deprecation warnings
* 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
2024-01-12 12:38:33 -05:00
Artem Bilan
f27a6410c6 Some Gradle build optimizations 2024-01-12 10:04:02 -05:00
Artem Bilan
e4ba493c94 Fix race condition in ObsPropagationChInterTests
The `QueueChannel.receive()` may answer before the `QueueChannel.send()`
observation is closed
2024-01-03 08:52:50 -05:00
Artem Bilan
23c792e11b Start version 6.3.0
* 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
2023-12-21 17:39:34 -05:00
Artem Bilan
5370d50932 GH-8773: Fix MGS for removal from group
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`**
2023-11-03 19:37:38 +01:00
Falk Hanisch
29186e2b46 GH-8757: Add nullabilty to .support.converter
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
2023-10-16 15:59:21 -04:00
Artem Bilan
6909fbe7ad Fix race condition in GatewayInterfaceTests 2023-10-16 15:34:14 -04:00
Artem Bilan
ea8d917278 Code clean up for UUIDConverter 2023-10-16 14:10:41 -04:00
Christian Tzolov
77d8dfd091 Streamline UUIDConverter
* 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`**
2023-10-16 14:02:58 -04:00
Johannes Edmeier
f89ca99bde Reduce log noise for AbstractCorrelatingMH
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`**
2023-10-16 11:04:47 -04:00
Falk Hanisch
49cd14cb8a GH-8750: Add @Nullable to getRegistrationById()
Fixes https://github.com/spring-projects/spring-integration/issues/8750

* annotate `IntegrationFlowRegistration.getRegistrationById(String flowId)` with `@org.springframework.lang.Nullable`
* annotate `StandardIntegrationFlowContext.getRegistrationById(String flowId)` with `@org.springframework.lang.Nullable`

**Cherry-pick to `6.1.x` & `6.0.x`**
2023-10-10 13:46:57 -04:00