diff --git a/README.md b/README.md index d21c5b5..d4bb319 100644 --- a/README.md +++ b/README.md @@ -3,6 +3,8 @@ > Distributed transaction library based on Choreography
+ + ![version 0.1.2](https://img.shields.io/badge/version-0.1.2-black?labelColor=black&style=flat-square) ![jdk 17](https://img.shields.io/badge/jdk-17-orange?labelColor=black&style=flat-square) @@ -38,17 +40,16 @@ class Application { #### Properties -| key | example | description | -|--------------------|-----------|------------------------------------------------------------------------------------------------------------------------------------| -| **netx.mode** | redis | 트랜잭션 관리에 사용할 메시지 큐 구현체의 mode 입니다. | -| **netx.host** | localhost | 트랜잭션 관리에 사용할 메시지 큐 의 host url 입니다. (ex. redis host) | -| **netx.port** | 6379 | 트랜잭션 관리에 사용할 메시지 큐의 port 입니다. | -| **netx.group** | pay-group | 분산 노드의 그룹입니다. 트랜잭션 이벤트는 같은 그룹내 하나의 노드로만 전송됩니다. | -| **netx.node-id** | 1 | id 생성에 사용될 식별자입니다. 모든 서버는 반드시 다른 id를 할당받아야 하며, 1~256 만큼의 id를 설정할 수 있습니다. _`중복된 id 생성을 방지하기위해 twitter snowflake 알고리즘으로 id를 생성합니다.`_ | -| **netx.node-name** | pay-1 | _`$netx.group`_ 에 참여할 서버의 이름입니다. 같은 그룹내에 중복된 이름이 존재하면 안됩니다. | -| **netx.undo.mode** | redis | 트랜잭션 undo 상태 저장에 사용할 저장소 구현체의 mode 입니다. | -| **netx.undo.host** | localhost | 트랜잭션 undo 상태 저장에 사용할 저장소의 host url 입니다. | -| **netx.undo.port** | 6380 | 트랜잭션 undo 상태 저장에 사용할 저장소의 port 입니다. | +| key | example | description | +|-------------------------|-----------|----------------------------------------------------------------------------------------------------------------------------------------------------| +| **netx.mode** | redis | 트랜잭션 관리에 사용할 메시지 큐 구현체의 mode 입니다. | +| **netx.host** | localhost | 트랜잭션 관리에 사용할 메시지 큐 의 host url 입니다. (ex. redis host) | +| **netx.port** | 6379 | 트랜잭션 관리에 사용할 메시지 큐의 port 입니다. | +| **netx.group** | pay-group | 분산 노드의 그룹입니다. 트랜잭션 이벤트는 같은 그룹내 하나의 노드로만 전송됩니다. | +| **netx.node-id** | 1 | id 생성에 사용될 식별자입니다. 모든 서버는 반드시 다른 id를 할당받아야 하며, 1~256 만큼의 id를 설정할 수 있습니다. _`중복된 id 생성을 방지하기위해 twitter snowflake 알고리즘으로 id를 생성합니다.`_ | +| **netx.node-name** | pay-1 | _`netx.group`_ 에 참여할 서버의 이름입니다. 같은 그룹내에 중복된 이름이 존재하면 안됩니다. | +| **netx.recovery-milli** | 60000 | _`netx.recovery-milli`_ 마다 _`netx.orphan-milli`_ 동안 처리 되지 않는 트랜잭션을 찾아 재실행합니다. 기본값은 60000(60초) 입니다. | +| **netx.orphan-milli** | 10000 | 트랜잭션이 PENDING 상태가 되었지만 orphan-milli가 지나도 ACK 상태가 되지 않는경우 다른 노드에게 처리를 위임합니다. 기본값은 10000(10초) 입니다. | ### Usage example diff --git a/build.gradle b/build.gradle index c643f27..e9761c0 100644 --- a/build.gradle +++ b/build.gradle @@ -35,7 +35,7 @@ publishing { } } -apply from: "gradle/mq.gradle" +apply from: "gradle/db.gradle" apply from: "gradle/test.gradle" apply from: "gradle/core.gradle" apply from: "gradle/sonar.gradle" diff --git a/gradle.properties b/gradle.properties index 0a8618e..a26f5c1 100644 --- a/gradle.properties +++ b/gradle.properties @@ -34,6 +34,9 @@ snowflakeVersion=5.2.5 ### Lettuce ### lettuceVersion=6.3.0.RELEASE +### Redisson ### +redissonVersion=3.26.0 + ### TestContainer ### testContainerVersion=1.19.3 diff --git a/gradle/mq.gradle b/gradle/db.gradle similarity index 56% rename from gradle/mq.gradle rename to gradle/db.gradle index d2ea2c7..f2cfa17 100644 --- a/gradle/mq.gradle +++ b/gradle/db.gradle @@ -1,3 +1,4 @@ dependencies { implementation "io.lettuce:lettuce-core:${lettuceVersion}" + implementation "org.redisson:redisson:${redissonVersion}" } diff --git a/idl b/idl index d0ea8e0..fdb6042 160000 --- a/idl +++ b/idl @@ -1 +1 @@ -Subproject commit d0ea8e0b64253d3e7966e2b6d95a751f40fe27de +Subproject commit fdb6042d5406ff052d07ca2997cf8f74c0608eaf diff --git a/src/main/kotlin/org/rooftop/netx/api/TransactionRollbackEvent.kt b/src/main/kotlin/org/rooftop/netx/api/TransactionRollbackEvent.kt index addfdce..63de5c6 100644 --- a/src/main/kotlin/org/rooftop/netx/api/TransactionRollbackEvent.kt +++ b/src/main/kotlin/org/rooftop/netx/api/TransactionRollbackEvent.kt @@ -4,5 +4,5 @@ data class TransactionRollbackEvent( val transactionId: String, val nodeName: String, val cause: String?, - val undoState: String, + val undo: String, ) diff --git a/src/main/kotlin/org/rooftop/netx/autoconfig/AutoConfigureDistributedTransaction.kt b/src/main/kotlin/org/rooftop/netx/autoconfig/AutoConfigureDistributedTransaction.kt index 093beb9..42278e7 100644 --- a/src/main/kotlin/org/rooftop/netx/autoconfig/AutoConfigureDistributedTransaction.kt +++ b/src/main/kotlin/org/rooftop/netx/autoconfig/AutoConfigureDistributedTransaction.kt @@ -1,10 +1,9 @@ package org.rooftop.netx.autoconfig import org.rooftop.netx.redis.RedisTransactionConfigurer -import org.rooftop.netx.redis.RedisUndoConfigurer import org.springframework.boot.autoconfigure.ImportAutoConfiguration @Target(AnnotationTarget.CLASS) @Retention(AnnotationRetention.RUNTIME) -@ImportAutoConfiguration(RedisUndoConfigurer::class, RedisTransactionConfigurer::class) +@ImportAutoConfiguration(RedisTransactionConfigurer::class) annotation class AutoConfigureDistributedTransaction diff --git a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt index 3450aac..92e40f5 100644 --- a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt +++ b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionDispatcher.kt @@ -6,39 +6,43 @@ import org.rooftop.netx.api.TransactionRollbackEvent import org.rooftop.netx.api.TransactionStartEvent import org.rooftop.netx.idl.Transaction import org.rooftop.netx.idl.TransactionState -import org.springframework.context.event.EventListener +import org.springframework.context.ApplicationEventPublisher import reactor.core.publisher.Flux import reactor.core.publisher.Mono abstract class AbstractTransactionDispatcher( - private val undoManager: UndoManager, - private val eventPublisher: EventPublisher, + private val eventPublisher: ApplicationEventPublisher, ) { - @EventListener(SubscribeTransactionEvent::class) - fun subscribeStream(event: SubscribeTransactionEvent): Flux { - return receive(event) - .dispatch() + fun subscribeStream(transactionId: String): Flux> { + return receive(transactionId) + .flatMap { dispatchAndAck(it.first, it.second) } } - protected abstract fun receive(event: SubscribeTransactionEvent): Flux + protected abstract fun receive(transactionId: String): Flux> - private fun Flux.dispatch(): Flux { - return this.flatMap { - when (it.state) { - TransactionState.TRANSACTION_STATE_JOIN -> publishJoin(it) - TransactionState.TRANSACTION_STATE_COMMIT -> publishCommit(it) - TransactionState.TRANSACTION_STATE_ROLLBACK -> publishRollback(it) - TransactionState.TRANSACTION_STATE_START -> publishStart(it) - else -> error("Cannot find matched transaction state \"${it.state}\"") - } + fun dispatchAndAck(transaction: Transaction, messageId: String): Flux> { + return Flux.just(transaction to messageId) + .dispatch() + .ack() + } + + private fun Flux>.dispatch(): Flux> { + return this.flatMap { (transaction, messageId) -> + when (transaction.state) { + TransactionState.TRANSACTION_STATE_JOIN -> publishJoin(transaction) + TransactionState.TRANSACTION_STATE_COMMIT -> publishCommit(transaction) + TransactionState.TRANSACTION_STATE_ROLLBACK -> publishRollback(transaction) + TransactionState.TRANSACTION_STATE_START -> publishStart(transaction) + else -> error("Cannot find matched transaction state \"${transaction.state}\"") + }.map { transaction to messageId } } } private fun publishJoin(it: Transaction): Mono { return Mono.just(it) .doOnNext { - eventPublisher.publish( + eventPublisher.publishEvent( TransactionJoinEvent( it.id, it.serverId @@ -49,29 +53,32 @@ abstract class AbstractTransactionDispatcher( private fun publishCommit(it: Transaction): Mono { return Mono.just(it) - .doOnNext { eventPublisher.publish(TransactionCommitEvent(it.id, it.serverId)) } + .doOnNext { eventPublisher.publishEvent(TransactionCommitEvent(it.id, it.serverId)) } } private fun publishRollback(transaction: Transaction): Mono { - return undoManager.find(transaction.id) + return findOwnTransaction(transaction) .doOnNext { - eventPublisher.publish( + eventPublisher.publishEvent( TransactionRollbackEvent( transaction.id, transaction.serverId, transaction.cause, - it + it.undo ) ) } - .flatMap { undoManager.delete(transaction.id) } .map { transaction } } + protected abstract fun findOwnTransaction(transaction: Transaction): Mono + private fun publishStart(it: Transaction): Mono { return Mono.just(it) .doOnNext { - eventPublisher.publish(TransactionStartEvent(it.id, it.serverId)) + eventPublisher.publishEvent(TransactionStartEvent(it.id, it.serverId)) } } + + protected abstract fun Flux>.ack(): Flux> } diff --git a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt index 0a77b6b..12a1f10 100644 --- a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt +++ b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionManager.kt @@ -5,59 +5,67 @@ import org.rooftop.netx.idl.Transaction import org.rooftop.netx.idl.TransactionState import org.rooftop.netx.idl.transaction import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers abstract class AbstractTransactionManager( nodeId: Int, + private val nodeGroup: String, private val nodeName: String, - private val eventPublisher: EventPublisher, private val transactionIdGenerator: TransactionIdGenerator = TransactionIdGenerator(nodeId), - private val undoManager: UndoManager, + private val transactionDispatcher: AbstractTransactionDispatcher, + private val transactionRetrySupporter: AbstractTransactionRetrySupporter, ) : TransactionManager { final override fun start(undo: String): Mono { - return startTransaction() + return startTransaction(undo) .subscribeTransaction() - .saveUndoState(undo) + .watchTransaction() .contextWrite { it.put(CONTEXT_TX_KEY, transactionIdGenerator.generate()) } } - private fun startTransaction(): Mono { + private fun startTransaction(undo: String): Mono { return Mono.deferContextual { Mono.just(it[CONTEXT_TX_KEY]) } .flatMap { transactionId -> publishTransaction(transactionId, transaction { id = transactionId serverId = nodeName + group = nodeGroup this.state = TransactionState.TRANSACTION_STATE_START + this.undo = undo }) } } final override fun join(transactionId: String, undo: String): Mono { return exists(transactionId) - .joinTransaction() + .joinTransaction(undo) .subscribeTransaction() - .saveUndoState(undo) + .watchTransaction() .contextWrite { it.put(CONTEXT_TX_KEY, transactionId) } } - private fun Mono.joinTransaction(): Mono { + private fun Mono.joinTransaction(undo: String): Mono { return flatMap { transactionId -> publishTransaction(transactionId, transaction { id = transactionId serverId = nodeName + group = nodeGroup state = TransactionState.TRANSACTION_STATE_JOIN + this.undo = undo }) } } private fun Mono.subscribeTransaction(): Mono { return this.doOnSuccess { - eventPublisher.publish(SubscribeTransactionEvent(it)) + transactionDispatcher.subscribeStream(it) + .subscribeOn(Schedulers.parallel()) + .subscribe() } } - private fun Mono.saveUndoState(undo: String): Mono { - return this.flatMap { undoManager.save(it, undo) } + private fun Mono.watchTransaction(): Mono { + return this.flatMap { transactionRetrySupporter.watchTransaction(it) } } final override fun rollback(transactionId: String, cause: String): Mono { @@ -65,6 +73,7 @@ abstract class AbstractTransactionManager( .publishTransaction(transaction { id = transactionId serverId = nodeName + group = nodeGroup state = TransactionState.TRANSACTION_STATE_ROLLBACK this.cause = cause }) @@ -76,6 +85,7 @@ abstract class AbstractTransactionManager( .publishTransaction(transaction { id = transactionId serverId = nodeName + group = nodeGroup state = TransactionState.TRANSACTION_STATE_COMMIT }) .contextWrite { it.put(CONTEXT_TX_KEY, transactionId) } diff --git a/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionRetrySupporter.kt b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionRetrySupporter.kt new file mode 100644 index 0000000..3f063c9 --- /dev/null +++ b/src/main/kotlin/org/rooftop/netx/engine/AbstractTransactionRetrySupporter.kt @@ -0,0 +1,24 @@ +package org.rooftop.netx.engine + +import org.rooftop.netx.idl.Transaction +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers +import kotlin.time.Duration.Companion.milliseconds +import kotlin.time.toJavaDuration + +abstract class AbstractTransactionRetrySupporter( + recoveryMilli: Long, +) { + + init { + Flux.interval(recoveryMilli.milliseconds.toJavaDuration()) + .publishOn(Schedulers.parallel()) + .flatMap { handleOrphanTransaction() } + .subscribe() + } + + abstract fun watchTransaction(transactionId: String): Mono + + protected abstract fun handleOrphanTransaction(): Flux> +} diff --git a/src/main/kotlin/org/rooftop/netx/engine/EventPublisher.kt b/src/main/kotlin/org/rooftop/netx/engine/EventPublisher.kt deleted file mode 100644 index c57d744..0000000 --- a/src/main/kotlin/org/rooftop/netx/engine/EventPublisher.kt +++ /dev/null @@ -1,6 +0,0 @@ -package org.rooftop.netx.engine - -fun interface EventPublisher { - - fun publish(event: Any) -} diff --git a/src/main/kotlin/org/rooftop/netx/engine/SubscribeTransactionEvent.kt b/src/main/kotlin/org/rooftop/netx/engine/SubscribeTransactionEvent.kt deleted file mode 100644 index 79e8447..0000000 --- a/src/main/kotlin/org/rooftop/netx/engine/SubscribeTransactionEvent.kt +++ /dev/null @@ -1,5 +0,0 @@ -package org.rooftop.netx.engine - -data class SubscribeTransactionEvent( - val transactionId: String -) diff --git a/src/main/kotlin/org/rooftop/netx/engine/UndoManager.kt b/src/main/kotlin/org/rooftop/netx/engine/UndoManager.kt deleted file mode 100644 index ab12d9e..0000000 --- a/src/main/kotlin/org/rooftop/netx/engine/UndoManager.kt +++ /dev/null @@ -1,13 +0,0 @@ -package org.rooftop.netx.engine - -import reactor.core.publisher.Mono - -interface UndoManager { - - fun save(transactionId: String, undo: String): Mono - - fun find(transactionId: String): Mono - - fun delete(transactionId: String): Mono - -} diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt index fc5e7d7..9d21605 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt @@ -1,9 +1,8 @@ package org.rooftop.netx.redis import org.rooftop.netx.engine.AbstractTransactionDispatcher -import org.rooftop.netx.engine.SubscribeTransactionEvent -import org.rooftop.netx.engine.UndoManager import org.rooftop.netx.idl.Transaction +import org.rooftop.netx.idl.TransactionState import org.springframework.context.ApplicationEventPublisher import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory import org.springframework.data.redis.connection.stream.Consumer @@ -12,6 +11,7 @@ import org.springframework.data.redis.connection.stream.StreamOffset import org.springframework.data.redis.core.ReactiveRedisTemplate import org.springframework.data.redis.stream.StreamReceiver import reactor.core.publisher.Flux +import reactor.core.publisher.Mono import reactor.core.scheduler.Schedulers import kotlin.time.Duration.Companion.hours import kotlin.time.toJavaDuration @@ -19,11 +19,10 @@ import kotlin.time.toJavaDuration class RedisStreamTransactionDispatcher( eventPublisher: ApplicationEventPublisher, connectionFactory: ReactiveRedisConnectionFactory, - undoManager: UndoManager, - private val streamGroup: String, + private val nodeGroup: String, private val nodeName: String, private val reactiveRedisTemplate: ReactiveRedisTemplate, -) : AbstractTransactionDispatcher(undoManager, SpringEventPublisher(eventPublisher)) { +) : AbstractTransactionDispatcher(eventPublisher) { private val options = StreamReceiver.StreamReceiverOptions.builder() .pollTimeout(1.hours.toJavaDuration()) @@ -31,20 +30,49 @@ class RedisStreamTransactionDispatcher( private val receiver = StreamReceiver.create(connectionFactory, options) - override fun receive(event: SubscribeTransactionEvent): Flux { - return createGroupIfNotExists(event) + override fun receive(transactionId: String): Flux> { + return createGroupIfNotExists(transactionId) .flatMap { - receiver.receiveAutoAck( - Consumer.from(streamGroup, nodeName), - StreamOffset.create(event.transactionId, ReadOffset.from(">")) + receiver.receive( + Consumer.from(nodeGroup, nodeName), + StreamOffset.create(transactionId, ReadOffset.from(">")) ).publishOn(Schedulers.parallel()) - .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) } + .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) to it.id.value } + .flatMap { (transaction, messageId) -> + when (transaction.state) { + TransactionState.TRANSACTION_STATE_ROLLBACK -> + findOwnTransaction(transaction).map { it to messageId } + + else -> Mono.just(transaction to messageId) + } + } } } - private fun createGroupIfNotExists(event: SubscribeTransactionEvent): Flux { + private fun createGroupIfNotExists(transactionId: String): Flux { return reactiveRedisTemplate.opsForStream() - .createGroup(event.transactionId, ReadOffset.from("0"), streamGroup) + .createGroup(transactionId, ReadOffset.from("0"), nodeGroup) .flatMapMany { Flux.just(it) } } + + override fun findOwnTransaction(transaction: Transaction): Mono { + return reactiveRedisTemplate.opsForStream() + .read(StreamOffset.create(transaction.id, ReadOffset.from("0"))) + .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) } + .filter { it.group == nodeGroup } + .filter { hasUndo(it) } + .next() + } + + private fun hasUndo(transaction: Transaction): Boolean = + transaction.state == TransactionState.TRANSACTION_STATE_JOIN + || transaction.state == TransactionState.TRANSACTION_STATE_START + + override fun Flux>.ack(): Flux> { + return this.flatMap { (transaction, messageId) -> + reactiveRedisTemplate.opsForStream() + .acknowledge(transaction.id, nodeGroup, messageId) + .flatMapMany { Flux.just(transaction to messageId) } + } + } } diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt index 2e37ef6..2a46bf8 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManager.kt @@ -1,9 +1,9 @@ package org.rooftop.netx.redis +import org.rooftop.netx.engine.AbstractTransactionDispatcher import org.rooftop.netx.engine.AbstractTransactionManager -import org.rooftop.netx.engine.UndoManager +import org.rooftop.netx.engine.AbstractTransactionRetrySupporter import org.rooftop.netx.idl.Transaction -import org.springframework.context.ApplicationEventPublisher import org.springframework.data.domain.Range import org.springframework.data.redis.connection.stream.Record import org.springframework.data.redis.core.ReactiveRedisTemplate @@ -12,14 +12,16 @@ import reactor.core.publisher.Mono class RedisStreamTransactionManager( nodeId: Int, nodeName: String, - applicationEventPublisher: ApplicationEventPublisher, - undoManager: UndoManager, + nodeGroup: String, + transactionDispatcher: AbstractTransactionDispatcher, + transactionRetrySupporter: AbstractTransactionRetrySupporter, private val reactiveRedisTemplate: ReactiveRedisTemplate, ) : AbstractTransactionManager( - nodeId, - nodeName, - SpringEventPublisher(applicationEventPublisher), - undoManager = undoManager + nodeId = nodeId, + nodeName = nodeName, + nodeGroup = nodeGroup, + transactionDispatcher = transactionDispatcher, + transactionRetrySupporter = transactionRetrySupporter, ) { override fun findAnyTransaction(transactionId: String): Mono { diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt index 376852f..b3b9030 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt @@ -1,7 +1,9 @@ package org.rooftop.netx.redis +import org.redisson.Redisson +import org.redisson.api.RedissonReactiveClient +import org.redisson.config.Config import org.rooftop.netx.api.TransactionManager -import org.rooftop.netx.engine.UndoManager import org.rooftop.pay.infra.transaction.ByteArrayRedisSerializer import org.springframework.beans.factory.annotation.Value import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty @@ -18,34 +20,48 @@ import org.springframework.data.redis.serializer.StringRedisSerializer class RedisTransactionConfigurer( @Value("\${netx.host}") private val host: String, @Value("\${netx.port}") private val port: String, - @Value("\${netx.group}") private val group: String, + @Value("\${netx.group}") private val nodeGroup: String, @Value("\${netx.node-id}") private val nodeId: Int, @Value("\${netx.node-name}") private val nodeName: String, + @Value("\${netx.recovery-milli:60000}") private val recoveryMilli: Long, + @Value("\${netx.orphan-milli:10000}") private val orphanMilli: Long, private val applicationEventPublisher: ApplicationEventPublisher, - private val undoManager: UndoManager, ) { @Bean @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") fun redisStreamTransactionManager(): TransactionManager = RedisStreamTransactionManager( - nodeId, - nodeName, - applicationEventPublisher, - undoManager, - reactiveRedisTemplate() + nodeId = nodeId, + nodeName = nodeName, + nodeGroup = nodeGroup, + transactionDispatcher = redisStreamTransactionDispatcher(), + transactionRetrySupporter = redisTransactionRetrySupporter(), + reactiveRedisTemplate = reactiveRedisTemplate(), ) @Bean @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") fun redisStreamTransactionDispatcher(): RedisStreamTransactionDispatcher = RedisStreamTransactionDispatcher( - applicationEventPublisher, - reactiveRedisConnectionFactory(), - undoManager, - group, - nodeName, - reactiveRedisTemplate() + eventPublisher = applicationEventPublisher, + connectionFactory = reactiveRedisConnectionFactory(), + nodeGroup = nodeGroup, + nodeName = nodeName, + reactiveRedisTemplate = reactiveRedisTemplate() + ) + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redisTransactionRetrySupporter(): RedisTransactionRetrySupporter = + RedisTransactionRetrySupporter( + nodeGroup = nodeGroup, + nodeName = nodeName, + reactiveRedisTemplate = reactiveRedisTemplate(), + redissonReactiveClient = redissonReactiveClient(), + transactionDispatcher = redisStreamTransactionDispatcher(), + orphanMilli = orphanMilli, + recoveryMilli = recoveryMilli, ) @Bean @@ -60,6 +76,18 @@ class RedisTransactionConfigurer( return ReactiveRedisTemplate(reactiveRedisConnectionFactory(), context) } + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redissonReactiveClient(): RedissonReactiveClient { + val port: String = System.getProperty("netx.port") ?: port + + return Redisson.create(Config() + .also { + it.useSingleServer() + .setAddress("redis://$host:$port") + }).reactive() + } + @Bean @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") fun byteArrayRedisSerializer(): ByteArrayRedisSerializer { diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt new file mode 100644 index 0000000..62373d7 --- /dev/null +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt @@ -0,0 +1,69 @@ +package org.rooftop.netx.redis + +import org.redisson.api.RedissonReactiveClient +import org.rooftop.netx.engine.AbstractTransactionDispatcher +import org.rooftop.netx.engine.AbstractTransactionRetrySupporter +import org.rooftop.netx.idl.Transaction +import org.springframework.data.domain.Range +import org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions +import org.springframework.data.redis.core.ReactiveRedisTemplate +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers +import java.util.concurrent.TimeUnit + +class RedisTransactionRetrySupporter( + private val nodeGroup: String, + private val nodeName: String, + private val reactiveRedisTemplate: ReactiveRedisTemplate, + private val redissonReactiveClient: RedissonReactiveClient, + private val transactionDispatcher: AbstractTransactionDispatcher, + private val orphanMilli: Long, + recoveryMilli: Long, +) : AbstractTransactionRetrySupporter(recoveryMilli) { + + override fun watchTransaction(transactionId: String): Mono { + return reactiveRedisTemplate.opsForSet() + .add(nodeGroup, transactionId.toByteArray()) + .map { transactionId } + } + + override fun handleOrphanTransaction(): Flux> { + return reactiveRedisTemplate.opsForSet() + .members(nodeGroup) + .flatMap { claimTransactions(String(it)) } + .publishOn(Schedulers.parallel()) + .flatMap { transactionDispatcher.dispatchAndAck(it.first, it.second) } + } + + private fun claimTransactions(transactionId: String): Flux> { + return reactiveRedisTemplate.opsForStream() + .pending(transactionId, nodeGroup, Range.closed("-", "+"), Long.MAX_VALUE) + .filter { it.get().toList().isNotEmpty() } + .flatMap { pendingMessage -> + redissonReactiveClient.getLock("$nodeGroup-key") + .tryLock(0, orphanMilli, TimeUnit.MILLISECONDS) + .map { pendingMessage } + } + .flatMapMany { + reactiveRedisTemplate.opsForStream() + .claim( + transactionId, nodeGroup, nodeName, XClaimOptions + .minIdleMs(orphanMilli) + .ids(it.get().map { eachMessage -> eachMessage.id.value }.toList()) + ) + } + .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) to it.id.toString() } + .flatMap { transactionWithMessageId -> + redissonReactiveClient.getLock("$nodeGroup-key") + .forceUnlock() + .flatMapMany { Flux.just(transactionWithMessageId) } + } + .doOnError { + redissonReactiveClient.getLock("$nodeGroup-key") + .unlock() + .subscribeOn(Schedulers.parallel()) + .subscribe() + } + } +} diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisUndoConfigurer.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisUndoConfigurer.kt deleted file mode 100644 index 5d807b2..0000000 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisUndoConfigurer.kt +++ /dev/null @@ -1,44 +0,0 @@ -package org.rooftop.netx.redis - -import org.springframework.beans.factory.annotation.Value -import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty -import org.springframework.context.annotation.Bean -import org.springframework.context.annotation.Configuration -import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory -import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory -import org.springframework.data.redis.core.ReactiveRedisTemplate -import org.springframework.data.redis.serializer.RedisSerializationContext -import org.springframework.data.redis.serializer.StringRedisSerializer - -@Configuration -class RedisUndoConfigurer( - @Value("\${netx.group}") private val group: String, - @Value("\${netx.undo.host}") private val netxUndoHost: String, - @Value("\${netx.undo.port}") private val netxUndoPort: String, -) { - - @Bean - @ConditionalOnProperty(prefix = "netx.undo", name = ["mode"], havingValue = "redis") - fun redisUndoManager(): RedisUndoManager = RedisUndoManager(group, redisUndoServer()) - - @Bean - @ConditionalOnProperty(prefix = "netx.undo", name = ["mode"], havingValue = "redis") - fun redisUndoServer(): ReactiveRedisTemplate { - val stringRedisSerializer = StringRedisSerializer() - - val context = - RedisSerializationContext.newSerializationContext(stringRedisSerializer) - .value(stringRedisSerializer) - .build() - - return ReactiveRedisTemplate(undoServerConnectionFactory(), context) - } - - @Bean - @ConditionalOnProperty(prefix = "netx.undo", name = ["mode"], havingValue = "redis") - fun undoServerConnectionFactory(): ReactiveRedisConnectionFactory { - val undoServerPort: String = System.getProperty("netx.undo.port") ?: netxUndoPort - - return LettuceConnectionFactory(netxUndoHost, undoServerPort.toInt()) - } -} diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisUndoManager.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisUndoManager.kt deleted file mode 100644 index 11d18c5..0000000 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisUndoManager.kt +++ /dev/null @@ -1,36 +0,0 @@ -package org.rooftop.netx.redis - -import org.rooftop.netx.engine.UndoManager -import org.springframework.data.redis.core.ReactiveRedisTemplate -import reactor.core.publisher.Mono - -class RedisUndoManager( - private val group: String, - private val reactiveRedisTemplate: ReactiveRedisTemplate, -) : UndoManager { - - override fun find(transactionId: String): Mono { - return reactiveRedisTemplate.opsForValue()["$group:$transactionId"] - .switchIfEmpty( - Mono.error { - throw IllegalStateException("Cannot find undo state \"$group:$transactionId\"") - } - ) - } - - override fun delete(transactionId: String): Mono { - return reactiveRedisTemplate.opsForValue().delete("$group:$transactionId") - } - - override fun save(transactionId: String, undo: String): Mono { - return reactiveRedisTemplate.opsForValue() - .set("$group:$transactionId", undo) - .flatMap { - when (it) { - true -> Mono.just(it) - false -> Mono.error { throw IllegalStateException("Error occurred during the undo process.") } - } - } - .map { transactionId } - } -} diff --git a/src/main/kotlin/org/rooftop/netx/redis/SpringEventPublisher.kt b/src/main/kotlin/org/rooftop/netx/redis/SpringEventPublisher.kt deleted file mode 100644 index 28ad8ad..0000000 --- a/src/main/kotlin/org/rooftop/netx/redis/SpringEventPublisher.kt +++ /dev/null @@ -1,12 +0,0 @@ -package org.rooftop.netx.redis - -import org.rooftop.netx.engine.EventPublisher -import org.rooftop.netx.engine.SubscribeTransactionEvent -import org.springframework.context.ApplicationEventPublisher - -class SpringEventPublisher(private val eventPublisher: ApplicationEventPublisher) : EventPublisher { - - override fun publish(event: Any) { - eventPublisher.publishEvent(event) - } -} diff --git a/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisStreamTransactionDispatcher.kt b/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisStreamTransactionDispatcher.kt new file mode 100644 index 0000000..6f79e54 --- /dev/null +++ b/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisStreamTransactionDispatcher.kt @@ -0,0 +1,73 @@ +package org.rooftop.netx.redis + +import org.rooftop.netx.engine.AbstractTransactionDispatcher +import org.rooftop.netx.idl.Transaction +import org.rooftop.netx.idl.TransactionState +import org.springframework.context.ApplicationEventPublisher +import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory +import org.springframework.data.redis.connection.stream.Consumer +import org.springframework.data.redis.connection.stream.ReadOffset +import org.springframework.data.redis.connection.stream.StreamOffset +import org.springframework.data.redis.core.ReactiveRedisTemplate +import org.springframework.data.redis.stream.StreamReceiver +import reactor.core.publisher.Flux +import reactor.core.publisher.Mono +import reactor.core.scheduler.Schedulers +import kotlin.time.Duration.Companion.hours +import kotlin.time.toJavaDuration + +class NoAckRedisStreamTransactionDispatcher( + eventPublisher: ApplicationEventPublisher, + connectionFactory: ReactiveRedisConnectionFactory, + private val nodeGroup: String, + private val nodeName: String, + private val reactiveRedisTemplate: ReactiveRedisTemplate, +) : AbstractTransactionDispatcher(eventPublisher) { + + private val options = StreamReceiver.StreamReceiverOptions.builder() + .pollTimeout(1.hours.toJavaDuration()) + .build() + + private val receiver = StreamReceiver.create(connectionFactory, options) + + override fun receive(transactionId: String): Flux> { + return createGroupIfNotExists(transactionId) + .flatMap { + receiver.receive( + Consumer.from(nodeGroup, nodeName), + StreamOffset.create(transactionId, ReadOffset.from(">")) + ).publishOn(Schedulers.parallel()) + .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) to it.id.value } + .flatMap { (transaction, messageId) -> + when (transaction.state) { + TransactionState.TRANSACTION_STATE_ROLLBACK -> + findOwnTransaction(transaction).map { it to messageId } + + else -> Mono.just(transaction to messageId) + } + } + } + } + + private fun createGroupIfNotExists(transactionId: String): Flux { + return reactiveRedisTemplate.opsForStream() + .createGroup(transactionId, ReadOffset.from("0"), nodeGroup) + .flatMapMany { Flux.just(it) } + } + + override fun findOwnTransaction(transaction: Transaction): Mono { + return reactiveRedisTemplate.opsForStream() + .read(StreamOffset.create(transaction.id, ReadOffset.from("0"))) + .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) } + .filter { it.group == nodeGroup } + .filter { hasUndo(it) } + .next() + } + + private fun hasUndo(transaction: Transaction): Boolean = + transaction.state == TransactionState.TRANSACTION_STATE_JOIN + || transaction.state == TransactionState.TRANSACTION_STATE_START + + override fun Flux>.ack(): Flux> = this +} + diff --git a/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt b/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt new file mode 100644 index 0000000..64d8cbe --- /dev/null +++ b/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt @@ -0,0 +1,117 @@ +package org.rooftop.netx.redis + +import org.redisson.Redisson +import org.redisson.api.RedissonReactiveClient +import org.redisson.config.Config +import org.rooftop.netx.api.TransactionManager +import org.rooftop.pay.infra.transaction.ByteArrayRedisSerializer +import org.springframework.beans.factory.annotation.Value +import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty +import org.springframework.boot.test.context.TestConfiguration +import org.springframework.context.ApplicationEventPublisher +import org.springframework.context.annotation.Bean +import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory +import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory +import org.springframework.data.redis.core.ReactiveRedisTemplate +import org.springframework.data.redis.serializer.RedisSerializationContext +import org.springframework.data.redis.serializer.StringRedisSerializer + +@TestConfiguration +class NoAckRedisTransactionConfigurer( + @Value("\${netx.host}") private val host: String, + @Value("\${netx.port}") private val port: String, + @Value("\${netx.group}") private val nodeGroup: String, + @Value("\${netx.node-id}") private val nodeId: Int, + @Value("\${netx.node-name}") private val nodeName: String, + @Value("\${netx.recovery-milli:60000}") private val recoveryMilli: Long, + @Value("\${netx.orphan-milli:10000}") private val orphanMilli: Long, + private val applicationEventPublisher: ApplicationEventPublisher, +) { + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redisStreamTransactionManager(): TransactionManager = + RedisStreamTransactionManager( + nodeId = nodeId, + nodeName = nodeName, + nodeGroup = nodeGroup, + transactionDispatcher = noAckRedisStreamTransactionDispatcher(), + transactionRetrySupporter = redisTransactionRetrySupporter(), + reactiveRedisTemplate = reactiveRedisTemplate(), + ) + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redisStreamTransactionDispatcher(): RedisStreamTransactionDispatcher = + RedisStreamTransactionDispatcher( + eventPublisher = applicationEventPublisher, + connectionFactory = reactiveRedisConnectionFactory(), + nodeGroup = nodeGroup, + nodeName = nodeName, + reactiveRedisTemplate = reactiveRedisTemplate() + ) + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun noAckRedisStreamTransactionDispatcher(): NoAckRedisStreamTransactionDispatcher = + NoAckRedisStreamTransactionDispatcher( + eventPublisher = applicationEventPublisher, + connectionFactory = reactiveRedisConnectionFactory(), + nodeGroup = nodeGroup, + nodeName = nodeName, + reactiveRedisTemplate = reactiveRedisTemplate() + ) + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redisTransactionRetrySupporter(): RedisTransactionRetrySupporter = + RedisTransactionRetrySupporter( + nodeGroup = nodeGroup, + nodeName = nodeName, + reactiveRedisTemplate = reactiveRedisTemplate(), + redissonReactiveClient = redissonReactiveClient(), + transactionDispatcher = redisStreamTransactionDispatcher(), + orphanMilli = orphanMilli, + recoveryMilli = recoveryMilli, + ) + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun reactiveRedisTemplate(): ReactiveRedisTemplate { + val builder = RedisSerializationContext.newSerializationContext( + StringRedisSerializer() + ) + + val context = builder.value(byteArrayRedisSerializer()).build() + + return ReactiveRedisTemplate(reactiveRedisConnectionFactory(), context) + } + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redissonReactiveClient(): RedissonReactiveClient { + val port: String = System.getProperty("netx.port") ?: port + + return Redisson.create( + Config() + .also { + it.useSingleServer() + .setAddress("redis://$host:$port") + }).reactive() + } + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun byteArrayRedisSerializer(): ByteArrayRedisSerializer { + return ByteArrayRedisSerializer() + } + + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun reactiveRedisConnectionFactory(): ReactiveRedisConnectionFactory { + val port: String = System.getProperty("netx.port") ?: port + + return LettuceConnectionFactory(host, port.toInt()) + } +} + diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisAssertions.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisAssertions.kt new file mode 100644 index 0000000..288bdb6 --- /dev/null +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisAssertions.kt @@ -0,0 +1,23 @@ +package org.rooftop.netx.redis + +import io.kotest.matchers.shouldBe +import org.springframework.beans.factory.annotation.Value +import org.springframework.boot.test.context.TestComponent +import org.springframework.data.domain.Range +import org.springframework.data.redis.core.ReactiveRedisOperations + +@TestComponent +internal class RedisAssertions( + private val reactiveRedisOperations: ReactiveRedisOperations, + @Value("\${netx.group}") private val nodeGroup: String, +) { + + fun pendingMessageCountShouldBe(transactionId: String, count: Long) { + val pendingMessageCount = reactiveRedisOperations.opsForStream() + .pending(transactionId, nodeGroup, Range.closed("-", "+"), Long.MAX_VALUE) + .map { it.get().toList().size } + .block() + + pendingMessageCount shouldBe count + } +} diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisContainer.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisContainer.kt index 8329778..1491200 100644 --- a/src/test/kotlin/org/rooftop/netx/redis/RedisContainer.kt +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisContainer.kt @@ -17,9 +17,5 @@ class RedisContainer { "netx.port", redis.getMappedPort(6379).toString() ) - System.setProperty( - "netx.undo.port", - redis.getMappedPort(6379).toString() - ) } } diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt index a81ecef..2c1906d 100644 --- a/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt @@ -131,6 +131,8 @@ internal class RedisStreamTransactionManagerTest( eventually(5.minutes) { eventCapture.capturedCount(TransactionRollbackEvent::class) } + + Thread.sleep(10.minutes.inWholeMilliseconds) } } diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporterTest.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporterTest.kt new file mode 100644 index 0000000..147575a --- /dev/null +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporterTest.kt @@ -0,0 +1,38 @@ +package org.rooftop.netx.redis + +import io.kotest.assertions.nondeterministic.eventually +import io.kotest.core.annotation.DisplayName +import io.kotest.core.spec.style.DescribeSpec +import org.rooftop.netx.api.TransactionManager +import org.springframework.test.context.ContextConfiguration +import org.springframework.test.context.TestPropertySource +import kotlin.time.Duration.Companion.minutes + +@ContextConfiguration( + classes = [ + RedisContainer::class, + RedisAssertions::class, + NoAckRedisTransactionConfigurer::class, + ] +) +@TestPropertySource("classpath:application.properties") +@DisplayName("RedisTransactionRetrySupporter 클래스의") +internal class RedisTransactionRetrySupporterTest( + private val redisAssertions: RedisAssertions, + private val transactionManager: TransactionManager, +) : DescribeSpec({ + + describe("handleOrphanTransaction 메소드는") { + context("pending되었지만, ack되지 않은 트랜잭션이 있다면,") { + it("해당 트랜잭션을 찾아서 처리하고, ack 상태로 변경한다.") { + val transactionId = transactionManager.start("undo").block()!! + + Thread.sleep(3_000) + + eventually(10.minutes) { + redisAssertions.pendingMessageCountShouldBe(transactionId, 0) + } + } + } + } +}) diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisUndoManagerTest.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisUndoManagerTest.kt deleted file mode 100644 index 268f976..0000000 --- a/src/test/kotlin/org/rooftop/netx/redis/RedisUndoManagerTest.kt +++ /dev/null @@ -1,77 +0,0 @@ -package org.rooftop.netx.redis - -import io.kotest.core.spec.style.DescribeSpec -import io.kotest.matchers.shouldBe -import org.rooftop.netx.autoconfig.AutoConfigureDistributedTransaction -import org.rooftop.netx.engine.UndoManager -import org.springframework.test.context.ContextConfiguration -import org.springframework.test.context.TestPropertySource -import reactor.test.StepVerifier - -@AutoConfigureDistributedTransaction -@ContextConfiguration(classes = [RedisContainer::class]) -@TestPropertySource("classpath:application.properties") -internal class RedisUndoManagerTest( - private val undoManager: UndoManager, -) : DescribeSpec({ - - describe("find 메소드는") { - context("transactionId를 받으면,") { - - val transactionId = "TX-1" - val undoState = "id:1" - undoManager.save(transactionId, undoState).block() - - it("저장된 UndoState를 반환한다.") { - val result = undoManager.find(transactionId) - - StepVerifier.create(result) - .assertNext { - it shouldBe undoState - } - .verifyComplete() - } - } - - context("존재하지 않는 transactionId를 받으면,") { - val transactionId = "UNKNOWN_TX" - - it("IllegalStateException 을 던진다.") { - val result = undoManager.find(transactionId) - - StepVerifier.create(result) - .verifyErrorMessage("Cannot find undo state \"netx-group:UNKNOWN_TX\"") - } - } - } - - describe("delete 메소드는") { - context("transactionId를 받으면,") { - - val transactionId = "TX-2" - val undoState = "id:2" - undoManager.save(transactionId, undoState).block() - - it("undoState 를 삭제하고 true를 반환한다.") { - val result = undoManager.delete(transactionId) - - StepVerifier.create(result) - .expectNext(true) - .verifyComplete() - } - } - - context("존재하지 않는 transactionId를 받으면") { - - val transactionId = "UNKNOWN_TX" - - it("false 를 반환한다.") { - val result = undoManager.delete(transactionId) - - StepVerifier.create(result) - .expectNext(false) - .verifyComplete() - } - } - } -}) diff --git a/src/test/resources/application.properties b/src/test/resources/application.properties index 91f67be..fdea166 100644 --- a/src/test/resources/application.properties +++ b/src/test/resources/application.properties @@ -4,6 +4,5 @@ netx.port=6379 netx.group=netx-group netx.node-id=1 netx.node-name=netx-node -netx.undo.mode=redis -netx.undo.host=localhost -netx.undo.port=6379 +netx.recovery-milli=1000 +netx.orphan-milli=1000