Although `RenewableLockRegistry` provides a renew interface, it is inconvenient for users.
Developers hope to have a lock that can be automatically renewed.
On the one hand, it can avoid subsequent failures caused by locks that will not expire when abnormal exits,
and on the other hand, it can avoid unlock failures caused by lock expired.
* Add `RenewableLockRegistry.setRenewalTaskScheduler()` and when it is set, schedule a `renew()` script periodically
when lock is acquired from Redis with `1/3` of `expireAfter`
* Test and document the feature
Fixes: #9401
Switched `tryLock()` with `lock()` in `RedisLockRegistry.runRedisMessageListenerContainer()` method for correct synchronization.
**Auto-cherry-pick to `6.3.x` & `6.2.x`**
Fixes: #9291
* Modify `unlock()` method of `JdbcLock`: if the lock ownership can not be removed due to data expiration, a `ConcurrentModificationException` should be thrown.
* Modify `unlock()` method of `RedisLock`: if the lock ownership can not be removed due to data expiration, a `ConcurrentModificationException` should be thrown.
* Maintain test cases
* Document the new behavior
Fixes: #8869
Do not set the `redisMessageListenerContainer`'s executor in the setter method
since the Redis Message Listener container is not initialised during configuration due to lazy loading
**Cherry-pick to `6.2.x` & `6.0.x`**
Fixes: #8803
Would be great to have an `Acquired lock; RedisLock...` logging message as we already have `Released lock; RedisLock ...`
* Improve `RedisLockRegistry.RedisLock.tryRedisLock()` to emit `Acquired lock; RedisLock...` debug message
**Cherry-pick to `6.0.x`**
The `ExpirableLockRegistry.expireUnusedOlderThan()` uses a query to the target store by mistake.
The logic of this API is indeed about the local cache for those lock instances.
We cannot effect any other processes with this cache.
And even if we remove our local instance while it is locked in other process,
that doesn't mean that on the next `obtain()` call, when we got a fresh local instance, we will be able to acquire a lock in target store.
* Fix `isAcquiredInThisProcess()` to check only `localLock.isLocked()`
**Cherry-pick to `6.1.x` & `6.0.x`**
In the current code, an `IllegalStateException` might be thrown from the try block while invoking the `removeLockKeyInnerUnlink()` method, especially when caused by key expiration (resulting in `unlinkResult == false`).
This triggers the check block, which incorrectly sets the `unlinkAvailable` flag to `false`, even if the Redis server supports the unlink operation.
As a consequence, the subsequent `removeLockKeyInnerDelete()` method is invoked when it should not be.
* `IllegalStateException` should not be thrown from try block
* Add a comment and fix Checkstyle violations
**Cherry-pick to `6.1.x` & `6.0.x`**
Expected Behavior
Using a single Lua script to verify ownership of the lock and remove it.
Current Behavior
`unlock()` method of `RedisLock` uses two separate Redis operations:
* `isAcquiredInThisProcess()`` method executes a `GET` operation to verify if the lock is owned by the process.
* `removeLockKey()`` method executes `UNLINK/DEL` operation to remove the lock.
* The `removeLockKeyInnerUnlink()`, and `removeLockKeyInnerDelete()`
methods will execute a script both verify ownership of the lock and remove it.
**Cherry-pick to `6.1.x` & `6.0.x`**
* Remove unused ThreadLocal from RedisStoreMSource
The `RedisStoreMessageSource` relies on an `IntegrationResourceHolder`
for a while already.
Its internal `ThreadLocal` is just out of use.
* Remove that `resourceHolder` property and related methods to interact with it.
* Mention the `IntegrationResourceHolder` and `store` attribute in the `redis.adoc`
* Fix language in Docs
Co-authored-by: Gary Russell <grussell@vmware.com>
---------
Co-authored-by: Gary Russell <grussell@vmware.com>
* 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
Currently, many timeouts in the project are like `-1` or other negative value
with a meaning to wait indefinitely.
According to distributed systems design and bad demo developing experience
it is not OK to block forever.
* Rework most of the timeouts in the framework to be `30` seconds.
Only one remained as `1` seconds is a `PollingConsumer` where it is
better to not block even for those 30 seconds when no messages in the queue,
but let the polling task be rescheduled.
* Remove the `MessagingGatewaySupport.replyTimeout` propagation down to the
`PollingConsumer` correlator where it was a `-1` before and blocked
the polling thread on the `Queue.poll()`.
This fixed the problem with a single thread in a pool for auto-configured `TaskScheduler`.
Now with 1 seconds wait time we are able to switch to other scheduled tasks
even with only 1 thread in the pool
Fix race condition, when methods `RedisLockRegistry#expireUnusedOlderThan` and `RedisLockRegistry#obtain` are executed successively.
It's possible to delete the lock from `RedisLockRegistry#expireUnusedOlderThan` method, when lock is created but is not acquired (`RedisLock#getLockedAt = 0`)
It can lead to the situation, when `RedisLockRegistry#obtain` returns multiple locks with the same redis-key, which shouldn't happen at all.
* Skip locks from expiration when their `lockedAt == 0` - new, not acquired yet.
**Cherry-pick to `6.0.x` & `5.5.x`**
Fixes https://github.com/spring-projects/spring-integration/issues/3686
* Add `src/idea` with respective editor config for IntelliJ IDEA.
Must be imported into an IDE
* Add `src/eclipse` with respective editor config for Eclipse/STS
* Reformat imports in source code according a new editor config
* GH-3888: Fix NPE in the `RedisLockRegistry.destroy()`
Fixed https://github.com/spring-projects/spring-integration/issues/3888
When `RedisLockType.SPIN_LOCK` (default), the `RedisLockRegistry.destroy()`
causes an NPE on the `redisMessageListenerContainer` since pub-sub is not used
in a busy-spin mode
* Check for `redisMessageListenerContainer` before calling its `destroy()`
**Cherry-pick to 5.5.x**
* * Reset properties in the `RedisLockRegistry` after `destroy()`
* Rework infra logic onto IntegrationProperties
The `Properties` bean has been deprecated since `5.5`
* Register a default `integrationGlobalProperties` as a `IntegrationProperties`
* Fix respective usages from the `Properties`
* * Remove unused constant from the `IntegrationContextUtils`
* * Fix language in the `whats-new.adoc`
Co-authored-by: Gary Russell <grussell@vmware.com>
Co-authored-by: Gary Russell <grussell@vmware.com>
* Fix deprecations around ListenableFuture
SF has deprecated a `ListenableFuture` and API around it
* Migrate to `CompletableFuture` everywhere a `ListenableFuture` has been used
* Suppress a deprecation for `ListenableFuture` keeping the functionality until the next version
* Resolve deprecations nad removals from the latest Spring for Apache Kafka
* Fix documentation for the `ListenableFuture` in favor of `CompletableFuture`
NOTE: the AMQP module is left as is until `ListenableFuture` deprecation is resolved in Spring AMQP
* * Restore some `ListenableFuture` test for messaging gateway
Fixes https://github.com/spring-projects/spring-integration/issues/3805
The Redis Pub-Sub doesn't work in all the environment, therefore there has to
be a choice to use old busy-spin algorithm
* Change to select between spinLock method and pub-sub method
* Make spinLock as a default one to let the `RedisLockRegistry` work everywhere
* Fix javadoc, convention, lazy init
* Fix javadoc, convention
* Code clean up and docs for `RedisLockType` feature
**Cherry-pick to 5.5.x**
Fixes https://github.com/spring-projects/spring-integration/issues/3716
If the `redisMessageListenerContainer` is starting, waiting for it to complete without doing `subscribeUnlock()`
* Introduce `isRunningRedisMessageListenerContainer` state since the `running` in the `RedisMessageListenerContainer` is set in the beginning of the `start()` misleading on the concurrent calls to the `RedisLockRegistry`
**Cherry-pick to `5.5.x`**
Fixes https://github.com/spring-projects/spring-integration/issues/3700
Fixed a case where lock acquisition attempt was abandoned earlier than expected time.
If a vm uses `tryLock()` and another vm unlock it, it will wake up with an unlock message and try to acquire the lock,
but if it fails(the other vm obtain lock), vm will return false.
Fixes https://github.com/spring-projects/spring-integration/issues/3439
The latest Spring Data Redis has introduced an `onErrorResume` function
option for the `StreamReceiver` and this one is now recommended way
to handle errors in the `Flux` from this receiver
* Expose all the `StreamReceiver.StreamReceiverOptionsBuilder` option
onto the `ReactiveRedisStreamMessageProducer`, including `onErrorResume`
* Have a default function as it was before - send into an error channel
supporting (n)ack in the failed message based on the failed record
* Make new setters mutually exclusive with an explicit `StreamReceiver.StreamReceiverOptions`
* Doc polishing
Fixes https://github.com/spring-projects/spring-integration/issues/3434
Each `RedisLock` has their own `unlinkAvailable` property
therefore we try to `unlink` on every unlock call and also log a WARN
about unavailability of this command in Redis, plus exception
* Move `unlinkAvailable` property to the `RedisLockRegistry` level
to have a check only once on a first `unlock()` call
* Move the whole exception logging onto the DEBUG level, leaving the WARN
only with the `ex.getMessage()`
* Optimize logging the same way in the `RedisMessageStore`
Cherry-pick to `5.3.x & 5.2.x`
Fixes https://github.com/spring-projects/spring-integration/issues/3428
The Kotlin Gradle plugin starting with Kotlin `1.4` adds a standard
Kotlin lib into `compile` scope fully ignoring an `optional` variant.
See more info here: https://kotlinlang.org/docs/reference/using-gradle.html#dependency-on-the-standard-library
* Add `kotlin.stdlib.default.dependency=false` into `gradle.properties`
to rely only on the explicit dependency definitions in the project config
* Remove explicit Kotlin deps from test scope in the target modules since
it is declared globally for all the modules
* Fix author emails in the `publish-maven.gradle` to their VMware variants
* Remove usage of `org.jetbrains.annotations` in favor of `@Nullable`
from Spring Framework
* Rely on `MProducerSupport.active` for `Flux`
* Fix `MessageProducerSupport` to extract an `active` flag and set it before
`isRunning` - the `Flux` subscription relies on the `takeWhile()`
where in case of `autoStartup = false` we will never start consume because
it is set to `true` already after `doStart()`
* Refactor all the `MessageProducerSupport` implementation with similar
`active` state to use already one from the super class
**Cherry-pick to 5.3.x**
* * Remove `MessageProducerSupport.setActive()`
to not let to mutate it from the implementations
* Set `active` to `false` in the `destroy()`
* Clean up and fix typos in the affected `JmsMessageDrivenEndpoint`
* * Pull `active` flag down to the `AbstractEndpoint`
* Set `active = true` in the `start()` before calling `doStart()`
* Do same for `active = false` in the `stop()`
* Clean up `AbstractEndpoint` impls to not call `doStart/doStop` for nothing
* Refactor endpoints to rely on the `active` state from the `AbstractEndpoint`
not their own
* Use 'onErrorContinue' to continue receiving messages from Stream
* Send an `ErrorMessage` to the provided `errorChannel` (if any)
* And `@Nullable` to some `MessageProducerSupport` API which definitely
may accept `null`
* Extract common `buildMessageFromRecord()` method in the `ReactiveRedisStreamMessageProducer`,
so all the headers from the stream `Record` are carried to the message independently of the
record state - normal send or error sending
The R2DBC channel adapters have wrong componentType
* Fix R2DBC component type to not mention `reactive-`: they are reactive by R2DBC definition
The `ReactiveRedisStreamMessageProducer` calls ack() explicitly
when populates `IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK` header,
but must provide an implementation of `SimpleAcknowledgment` instead
* Fix `ReactiveRedisStreamMessageProducer` to populate the proper
`IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK` header and in the proper moment
* Fix `ReactiveRedisStreamMessageProducerTests` to verify that `autoAck=false` works as expected
together with the `IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK` header functionality
