diff --git a/README.md b/README.md index d4bb319..30acdb9 100644 --- a/README.md +++ b/README.md @@ -40,16 +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.recovery-milli** | 60000 | _`netx.recovery-milli`_ 마다 _`netx.orphan-milli`_ 동안 처리 되지 않는 트랜잭션을 찾아 재실행합니다. 기본값은 60000(60초) 입니다. | -| **netx.orphan-milli** | 10000 | 트랜잭션이 PENDING 상태가 되었지만 orphan-milli가 지나도 ACK 상태가 되지 않는경우 다른 노드에게 처리를 위임합니다. 기본값은 10000(10초) 입니다. | +| 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/src/main/kotlin/org/rooftop/netx/redis/ByteArrayRedisSerializer.kt b/src/main/kotlin/org/rooftop/netx/redis/ByteArrayRedisSerializer.kt index 884998e..e8af044 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/ByteArrayRedisSerializer.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/ByteArrayRedisSerializer.kt @@ -1,4 +1,4 @@ -package org.rooftop.pay.infra.transaction +package org.rooftop.netx.redis import org.springframework.data.redis.serializer.RedisSerializer diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt index 9d21605..588587a 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionDispatcher.kt @@ -38,14 +38,6 @@ class RedisStreamTransactionDispatcher( 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) - } - } } } diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionRemover.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionRemover.kt new file mode 100644 index 0000000..8316bf1 --- /dev/null +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisStreamTransactionRemover.kt @@ -0,0 +1,53 @@ +package org.rooftop.netx.redis + +import org.rooftop.netx.api.TransactionCommitEvent +import org.rooftop.netx.api.TransactionRollbackEvent +import org.springframework.context.event.EventListener +import org.springframework.data.domain.Range +import org.springframework.data.redis.core.ReactiveRedisTemplate +import reactor.core.scheduler.Schedulers +import reactor.util.retry.RetrySpec +import java.time.Duration + +class RedisStreamTransactionRemover( + private val nodeGroup: String, + private val reactiveRedisTemplate: ReactiveRedisTemplate, +) { + + @EventListener(TransactionCommitEvent::class) + fun handleTransactionCommitEvent(event: TransactionCommitEvent) { + deleteElastic(event.transactionId) + } + + @EventListener(TransactionRollbackEvent::class) + fun handleTransactionRollbackEvent(event: TransactionRollbackEvent) { + deleteElastic(event.transactionId) + } + + private fun deleteElastic(transactionId: String) { + reactiveRedisTemplate.opsForStream() + .pending(transactionId, nodeGroup, Range.closed("-", "+"), Long.MAX_VALUE) + .filter { + when (it.get().toList().isEmpty()) { + true -> true + false -> error(TRANSACTION_IS_PENDING_STATUS) + } + } + .retryWhen(retryIfTransactionPending) + .flatMap { + reactiveRedisTemplate.opsForSet() + .remove(nodeGroup, transactionId.toByteArray()) + } + .subscribeOn(Schedulers.parallel()) + .subscribe() + } + + companion object { + private const val TRANSACTION_IS_PENDING_STATUS = + "Transaction message remains in pending status." + private val retryIfTransactionPending = + RetrySpec.fixedDelay(Long.MAX_VALUE, Duration.ofMillis(3000)) + .jitter(1.0) + .filter { it.message == TRANSACTION_IS_PENDING_STATUS } + } +} diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt index b3b9030..9a15925 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionConfigurer.kt @@ -4,7 +4,6 @@ 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.context.ApplicationEventPublisher @@ -64,6 +63,14 @@ class RedisTransactionConfigurer( recoveryMilli = recoveryMilli, ) + @Bean + @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") + fun redisStreamTransactionDeleter(): RedisStreamTransactionRemover = + RedisStreamTransactionRemover( + nodeGroup = nodeGroup, + reactiveRedisTemplate = reactiveRedisTemplate(), + ) + @Bean @ConditionalOnProperty(prefix = "netx", name = ["mode"], havingValue = "redis") fun reactiveRedisTemplate(): ReactiveRedisTemplate { diff --git a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt index 62373d7..065c433 100644 --- a/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt +++ b/src/main/kotlin/org/rooftop/netx/redis/RedisTransactionRetrySupporter.kt @@ -13,13 +13,14 @@ import reactor.core.scheduler.Schedulers import java.util.concurrent.TimeUnit class RedisTransactionRetrySupporter( + recoveryMilli: Long, 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, + private val lockKey: String = "$nodeGroup-key", ) : AbstractTransactionRetrySupporter(recoveryMilli) { override fun watchTransaction(transactionId: String): Mono { @@ -41,7 +42,7 @@ class RedisTransactionRetrySupporter( .pending(transactionId, nodeGroup, Range.closed("-", "+"), Long.MAX_VALUE) .filter { it.get().toList().isNotEmpty() } .flatMap { pendingMessage -> - redissonReactiveClient.getLock("$nodeGroup-key") + redissonReactiveClient.getLock(lockKey) .tryLock(0, orphanMilli, TimeUnit.MILLISECONDS) .map { pendingMessage } } @@ -55,12 +56,12 @@ class RedisTransactionRetrySupporter( } .map { Transaction.parseFrom(it.value["data"]?.toByteArray()) to it.id.toString() } .flatMap { transactionWithMessageId -> - redissonReactiveClient.getLock("$nodeGroup-key") + redissonReactiveClient.getLock(lockKey) .forceUnlock() .flatMapMany { Flux.just(transactionWithMessageId) } } .doOnError { - redissonReactiveClient.getLock("$nodeGroup-key") + redissonReactiveClient.getLock(lockKey) .unlock() .subscribeOn(Schedulers.parallel()) .subscribe() diff --git a/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt b/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt index 64d8cbe..7894e60 100644 --- a/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt +++ b/src/test/kotlin/org/rooftop/netx/redis/NoAckRedisTransactionConfigurer.kt @@ -4,7 +4,6 @@ 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 diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisAssertions.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisAssertions.kt index 288bdb6..de5c675 100644 --- a/src/test/kotlin/org/rooftop/netx/redis/RedisAssertions.kt +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisAssertions.kt @@ -20,4 +20,14 @@ internal class RedisAssertions( pendingMessageCount shouldBe count } + + fun retryTransactionShouldBeNotExists(transactionId: String) { + val retryTransaction = reactiveRedisOperations.opsForSet() + .members(nodeGroup) + .map { String(it) } + .any { it == transactionId } + .block() + + retryTransaction shouldBe false + } } diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt index 2c1906d..3cfa906 100644 --- a/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionManagerTest.kt @@ -129,10 +129,8 @@ internal class RedisStreamTransactionManagerTest( transactionManager.rollback(transactionId, "rollback occured for test").block() eventually(5.minutes) { - eventCapture.capturedCount(TransactionRollbackEvent::class) + eventCapture.capturedCount(TransactionRollbackEvent::class) shouldBe 1 } - - Thread.sleep(10.minutes.inWholeMilliseconds) } } diff --git a/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionRemoverTest.kt b/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionRemoverTest.kt new file mode 100644 index 0000000..4b9adef --- /dev/null +++ b/src/test/kotlin/org/rooftop/netx/redis/RedisStreamTransactionRemoverTest.kt @@ -0,0 +1,63 @@ +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 io.kotest.matchers.shouldBe +import org.rooftop.netx.api.TransactionCommitEvent +import org.rooftop.netx.api.TransactionManager +import org.rooftop.netx.api.TransactionRollbackEvent +import org.rooftop.netx.autoconfig.AutoConfigureDistributedTransaction +import org.springframework.test.context.ContextConfiguration +import org.springframework.test.context.TestPropertySource +import reactor.core.scheduler.Schedulers +import kotlin.time.Duration.Companion.seconds + +@AutoConfigureDistributedTransaction +@ContextConfiguration( + classes = [ + RedisContainer::class, + RedisAssertions::class, + EventCapture::class, + ] +) +@TestPropertySource("classpath:application.properties") +@DisplayName("RedisStreamTransactionRemover 클래스의") +internal class RedisStreamTransactionRemoverTest( + private val redisAssertions: RedisAssertions, + private val transactionManager: TransactionManager, + private val eventCapture: EventCapture, +) : DescribeSpec({ + describe("handleTransactionCommitEvent 메소드는") { + context("TransactionCommitEvent 가 발행되면,") { + val transactionId = transactionManager.start("RedisStreamTransactionRemoverTest") + .block()!! + + it("Transaction 을 retry watch 대기열에서 삭제한다.") { + transactionManager.commit(transactionId) + .subscribeOn(Schedulers.parallel()) + .subscribe() + + eventually(10.seconds) { + eventCapture.capturedCount(TransactionCommitEvent::class) shouldBe 1 + redisAssertions.retryTransactionShouldBeNotExists(transactionId) + } + } + } + + context("TransactionRollbackEvent 가 발행되면,") { + val transactionId = transactionManager.start("RedisStreamTransactionRemoverTest") + .block()!! + + it("Transaction 을 retry watch 대기열에서 삭제한다.") { + transactionManager.rollback(transactionId, "rollback occured for test").block() + + eventually(10.seconds) { + eventCapture.capturedCount(TransactionRollbackEvent::class) shouldBe 1 + redisAssertions.retryTransactionShouldBeNotExists(transactionId) + } + } + } + } +} +)