Commit Graph

2935 Commits

Author SHA1 Message Date
Artem Bilan
8f838d04ce GH-9455: Introduce IntegrationKeepAlive (#9493)
Fixes: #9455
Issue link: https://github.com/spring-projects/spring-integration/issues/9455

* Add an `IntegrationKeepAlive` infrastructure bean to initiate a long-lived non-daemon thread
to keep application alive when it cannot be kept like that for various reason, but has to.
* Expose `spring.integration.keepAlive` global property to disable an `IntegrationKeepAlive` auto-startup
* Test and document the feature
2024-09-25 14:31:41 -04:00
Artem Bilan
77ebee6d33 Fix Checkstyle violation 2024-09-24 17:25:52 -04:00
Artem Bilan
ec31a5bed8 GH-9507: Migrate Python support to GraalVM Polyglot
Fixes: #9507
Issue link: https://github.com/spring-projects/spring-integration/issues/9507

* Deprecate `PythonScriptExecutor` in favor of `PolyglotScriptExecutor` with a `python` as language
* Add handling for `PolyglotWrapper` return type of the script evaluation
* Rework `DeriveLanguageFromExtensionTests.testParseLanguage()` to the `@ParameterizedTest`
2024-09-24 17:19:43 -04:00
Artem Bilan
8082f3c3f9 Fix some compiler smells for Java 21 2024-09-24 14:33:16 -04:00
Tran Ngoc Nhan
fc377126de Modernize code for diamond, isEmpty & pattern matching 2024-09-23 14:42:38 -04:00
Artem Bilan
a8174d5bce GH-9478: Fix MessagingGatewaySupport.onInit() for calling super.onInit()
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

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
2024-09-17 12:16:31 -04:00
Artem Bilan
e3a46ca528 GH-9436: Add support for SpEL IndexAccessor configuration (#9451)
Fixes: #9436
Issue link: https://github.com/spring-projects/spring-integration/issues/9436

* Expose `IndexAccessor` configuration options on the `AbstractEvaluationContextFactoryBean`
and `SpelPropertyAccessorRegistrar`
* Expose `<index-accessors>` sub-element for the `<spel-property-accessors>`
* Adjust tests
* Document the feature, including recently added `JsonIndexAccessor`
2024-09-12 15:38:58 -04:00
Artem Bilan
775bfd6168 GH-9442: Register dynamic flows as singletons
Fixes: #9442
Issue link: https://github.com/spring-projects/spring-integration/issues/9442

When we register a `BeanDefinition`, its metadata is cached in the `DefaultListableBeanFactory.mergedBeanDefinitionHolders`
and it is not removed when we destroy bean.
This causes a memory leak with dynamic integration flows where even we destroy beans, their metadata is still cached.
This has more serious impact when random ids are used for dynamic integration flows.

* Rework `IntegrationFlowContext` logic to register singletons instead of `BeanDefinition`
* Adjust `IntegrationFlowBeanPostProcessor` logic to register singletons or `BeanDefinition`
according to the presence of `BeanDefinition` of the `IntegrationFlow` we are processing
* Introduce `ComponentSourceAware` to mimic a `BeanDefinition` metadata logic for bean source and its description.
This info helps a lot with exceptions where class in the `IntegrationFlow` might be fully from a Spring Integration package.
So, to give a clue what end-user code is related to the exception,
such a `ComponentSourceAware` is there to preserver and transfer "bean source"
* Implement `ComponentSourceAware` in the `IntegrationObjectSupport` since this is a central
place where we generate info for the Spring Integration exceptions
* Implement `ComponentSourceAware` in the `StandardIntegrationFlow` to propagate bean source info
down to its components for the mentioned `IntegrationObjectSupport` logic
* Introduce inner `StandardIntegrationFlowContext.IntegrationFlowComponentSourceAwareAdapter`
to be able to transfer "bean source" info from builder down to the target `IntegrationFlow` bean
2024-09-10 20:48:17 -04:00
Sam Brannen
49a0aaa793 GH-9383: Introduce JsonIndexAccessor
Fixes: #9383

Issue link: https://github.com/spring-projects/spring-integration/issues/9383

* Polish `JsonPropertyAccessor[Tests]`
* Introduce `JsonIndexAccessor`

This commit introduces a `JsonIndexAccessor` as a complement to the
existing `JsonPropertyAccessor`.

When a `JsonIndexAccessor` is registered with the SpEL `EvaluationContext`,
JSON arrays can be consistently indexed via integer literals (e.g.,[1]) instead of string literals representing integers (e.g., ['1']).
2024-09-04 10:25:46 -04:00
Artem Bilan
feef055a53 Use withAssignmentDisabled() for IntSimpEvalCtxFactoryBean
**Auto-cherry-pick to `6.3.x` & `6.2.x`**
2024-08-20 11:10:52 -04:00
Artem Bilan
b5ec098eb5 More @DirtiesContext for tests in core
An attempt to mitigate a memory impact on limited CI/CD runners

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
2024-08-13 16:33:47 -04:00
Artem Bilan
fdac8f1634 Fix tests for executorService.shutdown()
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.

**Auto-cherry-pick to `6.3.x` & `6.2.x`**
2024-08-13 15:57:25 -04:00
Artem Bilan
9180240ae0 Mitigate some IntegrationMBeanExporter early BF access
* Fix typo in the `spring-integration.xsd`
2024-08-13 14:15:42 -04:00
Artem Bilan
4d787554b8 GH-9381: Introduce Control Bus commands management
Fixes: #9381

Currently, there is no way to know in one place what Control Bus commands are available and with what arguments

* Add `ControlBusCommandRegistry` infrastructure bean to gather control bus commands from beans and expose them for invocation
* Add `ControlBusController` to expose a `/control-bus` REST service against the mentioned `ControlBusCommandRegistry`
* Add `@EnableIntegrationManagement(loadControlBusCommands)` to be able to load all the Control Bus commands from the application context instead of on demand by default
* Deprecated existing SpEL(and Groovy)-based Control Bus functionality in favor of new, more manageable, logic
2024-08-13 13:26:27 -04:00
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