Use separate transactions for implementations of Inbox and Transactional outbox patterns

rkudryashov · rkudryashov · commit 4b19483a0ce1 · 2024-09-06T16:03:57.000+03:00
This simplifies transaction management and error handling
diff --git a/book-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/bookservice/exception/Exceptions.kt b/book-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/bookservice/exception/Exceptions.kt
@@ -1,6 +1,6 @@
 package com.romankudryashov.eventdrivenarchitecture.bookservice.exception
 
-open class BookServiceException(message: String) : Exception(message)
+open class BookServiceException(message: String) : RuntimeException(message)
 
 class AccessRestrictedException : BookServiceException("Access to entity is restricted")
 
diff --git a/book-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/bookservice/service/impl/AuthorServiceImpl.kt b/book-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/bookservice/service/impl/AuthorServiceImpl.kt
@@ -35,7 +35,7 @@ class AuthorServiceImpl(
 
     override fun getEntityById(id: Long): AuthorEntity? = authorRepository.findByIdOrNull(id)
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])
     override fun update(id: Long, author: AuthorToSave): AuthorDto {
         log.debug("Start update an author: id={}, new state={}", id, author)
 
diff --git a/book-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/bookservice/service/impl/BookServiceImpl.kt b/book-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/bookservice/service/impl/BookServiceImpl.kt
@@ -50,7 +50,7 @@ class BookServiceImpl(
         bookEntityToDtoConverter.convert(entity)
     }
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])
     override fun create(book: BookToSave): BookDto {
         log.debug("Start creating a book: {}", book)
 
@@ -63,7 +63,7 @@ class BookServiceImpl(
         return bookEntityToDtoConverter.convert(createdBook)
     }
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])
     override fun update(id: Long, book: BookToSave): BookDto {
         log.debug("Start updating a book: id={}, new state={}", id, book)
 
@@ -78,7 +78,7 @@ class BookServiceImpl(
         return bookEntityToDtoConverter.convert(updatedBook)
     }
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])
     override fun delete(id: Long) {
         log.debug("Start deleting a book: id={}", id)
 
@@ -94,7 +94,7 @@ class BookServiceImpl(
         }
     }
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])
     override fun lendBook(bookId: Long, bookLoan: BookLoanToSave): BookLoanDto {
         log.debug("Start lending a book: bookId={}, bookLoan={}", bookId, bookLoan)
 
@@ -116,7 +116,7 @@ class BookServiceImpl(
         return bookLoanEntityToDtoConverter.convert(bookToLend.currentLoan()!!)
     }
 
-    @Transactional(propagation = Propagation.MANDATORY, noRollbackFor = [RuntimeException::class])
+    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])
     override fun cancelBookLoan(bookId: Long, bookLoanId: Long) {
         log.debug("Start cancelling book loan for a book: bookId={}, bookLoanId={}", bookId, bookLoanId)
 
@@ -131,7 +131,7 @@ class BookServiceImpl(
         outboxMessageService.saveBookLoanCanceledEventMessage(modelOfBookToCancelLoan)
     }
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])
     override fun returnBook(bookId: Long, bookLoanId: Long) {
         log.debug("Start returning a book: bookId={}, bookLoanId={}", bookId, bookLoanId)
 
diff --git a/book-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/bookservice/service/impl/InboxMessageServiceImpl.kt b/book-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/bookservice/service/impl/InboxMessageServiceImpl.kt
@@ -22,7 +22,7 @@ class InboxMessageServiceImpl(
 
     private val log = LoggerFactory.getLogger(this.javaClass)
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    @Transactional(propagation = Propagation.REQUIRES_NEW, noRollbackFor = [RuntimeException::class])
     override fun markInboxMessagesAsReadyForProcessingByInstance(batchSize: Int): Int {
         fun saveReadyForProcessing(inboxMessage: InboxMessageEntity) {
             log.debug("Start saving a message ready for processing, id={}", inboxMessage.id)
@@ -46,7 +46,7 @@ class InboxMessageServiceImpl(
     override fun getBatchForProcessing(batchSize: Int): List<InboxMessageEntity> =
         inboxMessageRepository.findAllByStatusAndProcessedByOrderByCreatedAtAsc(InboxMessageEntity.Status.ReadyForProcessing, applicationName, PageRequest.of(0, batchSize))
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    @Transactional(propagation = Propagation.REQUIRES_NEW, noRollbackFor = [RuntimeException::class])
     override fun process(inboxMessage: InboxMessageEntity) {
         log.debug("Start processing an inbox message with id={}", inboxMessage.id)
 
diff --git a/book-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/bookservice/service/impl/IncomingEventServiceImpl.kt b/book-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/bookservice/service/impl/IncomingEventServiceImpl.kt
@@ -22,7 +22,7 @@ class IncomingEventServiceImpl(
 
     private val log = LoggerFactory.getLogger(this.javaClass)
 
-    @Transactional(propagation = Propagation.MANDATORY, noRollbackFor = [RuntimeException::class])
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     override fun process(eventType: EventType, payload: JsonNode) {
         log.debug("Start processing an incoming event: type={}, payload={}", eventType, payload)
 
diff --git a/notification-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/notificationservice/exception/Exceptions.kt b/notification-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/notificationservice/exception/Exceptions.kt
@@ -1,3 +1,3 @@
 package com.romankudryashov.eventdrivenarchitecture.notificationservice.exception
 
-class NotificationServiceException(message: String) : Exception(message)
+class NotificationServiceException(message: String) : RuntimeException(message)
diff --git a/notification-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/notificationservice/service/impl/InboxMessageServiceImpl.kt b/notification-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/notificationservice/service/impl/InboxMessageServiceImpl.kt
@@ -22,7 +22,7 @@ class InboxMessageServiceImpl(
 
     private val log = LoggerFactory.getLogger(this.javaClass)
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    @Transactional(propagation = Propagation.REQUIRES_NEW, noRollbackFor = [RuntimeException::class])
     override fun markInboxMessagesAsReadyForProcessingByInstance(batchSize: Int): Int {
         fun saveReadyForProcessing(inboxMessage: InboxMessageEntity) {
             log.debug("Start saving a message ready for processing, id={}", inboxMessage.id)
@@ -46,7 +46,7 @@ class InboxMessageServiceImpl(
     override fun getBatchForProcessing(batchSize: Int): List<InboxMessageEntity> =
         inboxMessageRepository.findAllByStatusAndProcessedByOrderByCreatedAtAsc(InboxMessageEntity.Status.ReadyForProcessing, applicationName, PageRequest.of(0, batchSize))
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    @Transactional(propagation = Propagation.REQUIRES_NEW, noRollbackFor = [RuntimeException::class])
     override fun process(inboxMessage: InboxMessageEntity) {
         log.debug("Start processing an inbox message with id={}", inboxMessage.id)
 
diff --git a/notification-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/notificationservice/service/impl/IncomingEventServiceImpl.kt b/notification-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/notificationservice/service/impl/IncomingEventServiceImpl.kt
@@ -22,7 +22,7 @@ class IncomingEventServiceImpl(
 
     private val log = LoggerFactory.getLogger(this.javaClass)
 
-    @Transactional(propagation = Propagation.MANDATORY, noRollbackFor = [RuntimeException::class])
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     override fun process(eventType: EventType, payload: JsonNode) {
         log.debug("Start processing an incoming event: type={}, payload={}", eventType, payload)
 
diff --git a/user-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/userservice/exception/Exceptions.kt b/user-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/userservice/exception/Exceptions.kt
@@ -1,3 +1,3 @@
 package com.romankudryashov.eventdrivenarchitecture.userservice.exception
 
-class UserServiceException(message: String) : Exception(message)
+class UserServiceException(message: String) : RuntimeException(message)
diff --git a/user-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/userservice/service/impl/InboxMessageServiceImpl.kt b/user-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/userservice/service/impl/InboxMessageServiceImpl.kt
@@ -22,7 +22,7 @@ class InboxMessageServiceImpl(
 
     private val log = LoggerFactory.getLogger(this.javaClass)
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    @Transactional(propagation = Propagation.REQUIRES_NEW, noRollbackFor = [RuntimeException::class])
     override fun markInboxMessagesAsReadyForProcessingByInstance(batchSize: Int): Int {
         fun saveReadyForProcessing(inboxMessage: InboxMessageEntity) {
             log.debug("Start saving a message ready for processing, id={}", inboxMessage.id)
@@ -46,7 +46,7 @@ class InboxMessageServiceImpl(
     override fun getBatchForProcessing(batchSize: Int): List<InboxMessageEntity> =
         inboxMessageRepository.findAllByStatusAndProcessedByOrderByCreatedAtAsc(InboxMessageEntity.Status.ReadyForProcessing, applicationName, PageRequest.of(0, batchSize))
 
-    @Transactional(propagation = Propagation.REQUIRES_NEW)
+    @Transactional(propagation = Propagation.REQUIRES_NEW, noRollbackFor = [RuntimeException::class])
     override fun process(inboxMessage: InboxMessageEntity) {
         log.debug("Start processing an inbox message with id={}", inboxMessage.id)
 
diff --git a/user-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/userservice/service/impl/IncomingEventServiceImpl.kt b/user-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/userservice/service/impl/IncomingEventServiceImpl.kt
@@ -38,7 +38,7 @@ class IncomingEventServiceImpl(
 
     private val log = LoggerFactory.getLogger(this.javaClass)
 
-    @Transactional(propagation = Propagation.MANDATORY, noRollbackFor = [RuntimeException::class])
+    @Transactional(propagation = Propagation.NOT_SUPPORTED)
     override fun process(eventType: EventType, payload: JsonNode) {
         log.debug("Start processing an incoming event: type={}, payload={}", eventType, payload)
 
diff --git a/user-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/userservice/service/impl/OutboxMessageServiceImpl.kt b/user-service/src/main/kotlin/com/romankudryashov/eventdrivenarchitecture/userservice/service/impl/OutboxMessageServiceImpl.kt
@@ -14,6 +14,8 @@ import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.ObjectMapper
 import org.slf4j.LoggerFactory
 import org.springframework.stereotype.Service
+import org.springframework.transaction.annotation.Propagation
+import org.springframework.transaction.annotation.Transactional
 
 @Service
 class OutboxMessageServiceImpl(
@@ -28,8 +30,10 @@ class OutboxMessageServiceImpl(
         RollbackBookLentCommand to "rollback"
     )
 
+    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])
     override fun saveRollbackBookLentCommandMessage(payload: Book) = save(createOutboxMessage(AggregateType.Book, payload.id, RollbackBookLentCommand, payload))
 
+    @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])
     override fun saveSendNotificationCommandMessage(payload: Notification, aggregateId: Long) =
         save(createOutboxMessage(AggregateType.Notification, null, SendNotificationCommand, payload))
 

Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,7 @@ class BookServiceImpl(`
`50`	`50`	`bookEntityToDtoConverter.convert(entity)`
`51`	`51`	`}`
`52`	`52`
`53`		`- @Transactional(propagation = Propagation.REQUIRES_NEW)`
	`53`	`+ @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])`
`54`	`54`	`override fun create(book: BookToSave): BookDto {`
`55`	`55`	`log.debug("Start creating a book: {}", book)`
`56`	`56`
`@@ -63,7 +63,7 @@ class BookServiceImpl(`
`63`	`63`	`return bookEntityToDtoConverter.convert(createdBook)`
`64`	`64`	`}`
`65`	`65`
`66`		`- @Transactional(propagation = Propagation.REQUIRES_NEW)`
	`66`	`+ @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])`
`67`	`67`	`override fun update(id: Long, book: BookToSave): BookDto {`
`68`	`68`	`log.debug("Start updating a book: id={}, new state={}", id, book)`
`69`	`69`
`@@ -78,7 +78,7 @@ class BookServiceImpl(`
`78`	`78`	`return bookEntityToDtoConverter.convert(updatedBook)`
`79`	`79`	`}`
`80`	`80`
`81`		`- @Transactional(propagation = Propagation.REQUIRES_NEW)`
	`81`	`+ @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])`
`82`	`82`	`override fun delete(id: Long) {`
`83`	`83`	`log.debug("Start deleting a book: id={}", id)`
`84`	`84`
`@@ -94,7 +94,7 @@ class BookServiceImpl(`
`94`	`94`	`}`
`95`	`95`	`}`
`96`	`96`
`97`		`- @Transactional(propagation = Propagation.REQUIRES_NEW)`
	`97`	`+ @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])`
`98`	`98`	`override fun lendBook(bookId: Long, bookLoan: BookLoanToSave): BookLoanDto {`
`99`	`99`	`log.debug("Start lending a book: bookId={}, bookLoan={}", bookId, bookLoan)`
`100`	`100`
`@@ -116,7 +116,7 @@ class BookServiceImpl(`
`116`	`116`	`return bookLoanEntityToDtoConverter.convert(bookToLend.currentLoan()!!)`
`117`	`117`	`}`
`118`	`118`
`119`		`- @Transactional(propagation = Propagation.MANDATORY, noRollbackFor = [RuntimeException::class])`
	`119`	`+ @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])`
`120`	`120`	`override fun cancelBookLoan(bookId: Long, bookLoanId: Long) {`
`121`	`121`	`log.debug("Start cancelling book loan for a book: bookId={}, bookLoanId={}", bookId, bookLoanId)`
`122`	`122`
`@@ -131,7 +131,7 @@ class BookServiceImpl(`
`131`	`131`	`outboxMessageService.saveBookLoanCanceledEventMessage(modelOfBookToCancelLoan)`
`132`	`132`	`}`
`133`	`133`
`134`		`- @Transactional(propagation = Propagation.REQUIRES_NEW)`
	`134`	`+ @Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = [Exception::class])`
`135`	`135`	`override fun returnBook(bookId: Long, bookLoanId: Long) {`
`136`	`136`	`log.debug("Start returning a book: bookId={}, bookLoanId={}", bookId, bookLoanId)`
`137`	`137`
Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,3 @@`
`1`	`1`	`package com.romankudryashov.eventdrivenarchitecture.notificationservice.exception`
`2`	`2`
`3`		`-class NotificationServiceException(message: String) : Exception(message)`
	`3`	`+class NotificationServiceException(message: String) : RuntimeException(message)`
Original file line number	Diff line number	Diff line change
`@@ -1,3 +1,3 @@`
`1`	`1`	`package com.romankudryashov.eventdrivenarchitecture.userservice.exception`
`2`	`2`
`3`		`-class UserServiceException(message: String) : Exception(message)`
	`3`	`+class UserServiceException(message: String) : RuntimeException(message)`