StackSaga-Kafka Framework (Asynchronous)

@Component (1)
public class OrderDomainEntityMapperProvider extends AbstractDomainEntityMapperProvider { (2)

    @Override (3)
    protected ObjectMapper provide() {
        return new ObjectMapper(); (4)
    }
}
//-------------------------------------------------------------------------------

@Getter
@Setter
@SagaDomainEntity(
        version = @SagaDomainEntityVersion(major = 1, minor = 0, patch = 0),
        name = "OrderDomainEntity",
        mapper = OrderDomainEntityMapperProvider.class (5)
)
public class OrderDomainEntity extends DomainEntity {
    //...
}

@Component (1)
public class OrderDomainEntityKeyGenerator extends AbstractDomainEntityKeyGenerator { (2)

    @Override (3)
    // this method is called when each transaction is initialized
    public String generateTransactionKey(String serviceName, String applicationVersion, String instanceId, String region, String zone, SagaDomainEntity sagaDomainEntity) {

        StringBuilder regionKey = new StringBuilder();
        for (char c : region.toCharArray()) {
            regionKey.append((int) c);
        }
        return String.format("%s-%s-%s", serviceName , regionKey , UUID.randomUUID());
    }

    @Override (4)
    public String generateIdempotencyKey(SafeIdempotentInput safeIdempotentInput, UnSafeIdempotentInput unSafeIdempotentInput) {
         final String rowKey = new StringJoiner(":")
                .add(safeIdempotentInput.transactionId())
                .add(safeIdempotentInput.currentExecutionName())
                .add(safeIdempotentInput.executionMode().name().toLowerCase())
                .toString();
       return this.hashGenerator.generateHash(rowKey, HashGenerator.ALGType.MD5);
    }
}

//: Configure The custom KeyGen With Custom DomainEntity

@Getter
@Setter
@SagaDomainEntity(
        version = @SagaDomainEntityVersion(major = 1, minor = 0, patch = 0),
        name = "OrderDomainEntity",
        mapper = OrderDomainEntityMapperProvider.class,
        keyGen = OrderDomainEntityKeyGenerator.class (5)
)
public class OrderDomainEntity extends DomainEntity {
    //...
}

class PlaceOrderTopic extends AbstractTopic<PlaceOrderTopic> {(1)

    (2)
    protected PlaceOrderTopic(String topicName, float topicKey, SagaEventType sagaEventType, String targetService) {
        super(topicName, topicKey, sagaEventType, targetService);
    }


    (3)
    protected PlaceOrderTopic(String topicName, float topicKey, SagaEventType sagaEventType, String targetService, PlaceOrderTopic parent) {
        super(topicName, topicKey, sagaEventType, targetService, parent);
    }

    (4)
    //primary execution topics.
    public static final PlaceOrderTopic DO_FETCH_USER_DETAILS = new PlaceOrderTopic("user.fetch", 1, SagaEventType.QUERY_DO_ACTION, "user-service");
    public static final PlaceOrderTopic DO_INITIALIZE_ORDER = new PlaceOrderTopic("order.init", 2, SagaEventType.COMMAND_DO_ACTION, "order-service");
    public static final PlaceOrderTopic DO_MAKE_PAYMENT = new PlaceOrderTopic("payment.make", 3, SagaEventType.COMMAND_DO_ACTION, "payment-service");
    public static final PlaceOrderTopic DO_INVENTORY_UPDATE = new PlaceOrderTopic("inventory.update", 4, SagaEventType.COMMAND_DO_ACTION, "inventory-service");

    (5)
    //revert/compensation topics.
    public static final PlaceOrderTopic UNDO_INITIALIZE_ORDER = new PlaceOrderTopic("order.init", -2, SagaEventType.COMMAND_UNDO_ACTION, "order-service", DO_INITIALIZE_ORDER);
    public static final PlaceOrderTopic UNDO_MAKE_PAYMENT = new PlaceOrderTopic("payment.make", -3, SagaEventType.COMMAND_UNDO_ACTION, "payment-service", DO_MAKE_PAYMENT);
    public static final PlaceOrderTopic UNDO_INVENTORY_UPDATE = new PlaceOrderTopic("inventory.update", -4, SagaEventType.COMMAND_UNDO_ACTION, "inventory-service", DO_INVENTORY_UPDATE);
}

(2)
@SagaEventManager(
        value = "placeOrderEventManager", (3)
        listenerScope = ListenerScope.SHARED, (4)
        domainRootTopicSuffix = "place-order" (5)
)
public class PlaceOrderEventManager extends AbstractEventManager<OrderDomainEntity, PlaceOrderTopic> { (1)

    (6)
    @Override
    public Supplier<List<PlaceOrderTopic>> registerTopics() {
        return () -> List.of(
                PlaceOrderTopic.DO_FETCH_USER_DETAILS,
                PlaceOrderTopic.DO_INITIALIZE_ORDER,
                PlaceOrderTopic.DO_MAKE_PAYMENT,
                PlaceOrderTopic.DO_INVENTORY_UPDATE,
                PlaceOrderTopic.UNDO_INITIALIZE_ORDER,
                PlaceOrderTopic.UNDO_MAKE_PAYMENT,
                PlaceOrderTopic.UNDO_INVENTORY_UPDATE
        );
    }

    (7)
    @Override
    public @NonNull SagaPrimaryEventAction<PlaceOrderTopic> onNext(PlaceOrderTopic recentTopic, OrderDomainEntity currentDomainEntityState) {
        if (recentTopic.equals(PlaceOrderTopic.DO_FETCH_USER_DETAILS)) {
            currentDomainEntityState.getMetadata().put("userDetailsFetched", "true");
            return SagaPrimaryEventAction.next(PlaceOrderTopic.DO_INITIALIZE_ORDER);
        }
        if (recentTopic.equals(PlaceOrderTopic.DO_INITIALIZE_ORDER)) {
            currentDomainEntityState.getMetadata().put("orderInitialized", "true");
            return SagaPrimaryEventAction.next(PlaceOrderTopic.DO_MAKE_PAYMENT);
        }
        if (recentTopic.equals(PlaceOrderTopic.DO_MAKE_PAYMENT)) {
            currentDomainEntityState.getMetadata().put("paymentMade", "true");
            return SagaPrimaryEventAction.next(PlaceOrderTopic.DO_INVENTORY_UPDATE);
        }
        if (recentTopic.equals(PlaceOrderTopic.DO_INVENTORY_UPDATE)) {
            currentDomainEntityState.getMetadata().put("inventoryUpdated", "true");
            return SagaPrimaryEventAction.complete();
        }
        return SagaPrimaryEventAction.error(new IllegalStateException("Unexpected topic: " + recentTopic));
    }

    (8)
    @Override
    public void onNextRevert(
            PlaceOrderTopic recentExecutedTopic,
            PlaceOrderTopic nextTopic,
            OrderDomainEntity lastDomainEntityState,
            NonRetryableExecutorException nonRetryableExecutorException,
            RevertHintStore revertHintStore,
            Supplier<NavigableMap<Integer, PlaceOrderTopic>> remainingReverts
    ) {

        (9)
        {//sample usage of revert hint store and next topic in the onNextRevert method.
            if (nextTopic.equals(PlaceOrderTopic.UNDO_MAKE_PAYMENT)) {
                revertHintStore.put("BEFORE_NOTE:UNDO_MAKE_PAYMENT", "Sample value before reverting UNDO_MAKE_PAYMENT");
            }
            if (nextTopic.equals(PlaceOrderTopic.UNDO_INITIALIZE_ORDER)) {
                revertHintStore.put("BEFORE_NOTE:UNDO_INITIALIZE_ORDER", "Sample value before reverting UNDO_INITIALIZE_ORDER");
            }
        }

        (10)
        {//sample usage of remaining reverts and recentExecutedTopic
            if (recentExecutedTopic.equals(PlaceOrderTopic.UNDO_MAKE_PAYMENT)) {
                log.info("Remaining reverts after reverting UNDO_MAKE_PAYMENT: {}", remainingReverts);
            }

            if (recentExecutedTopic.equals(PlaceOrderTopic.UNDO_INITIALIZE_ORDER)) {
                log.info("Remaining reverts after reverting UNDO_INITIALIZE_ORDER: {}", remainingReverts);
            }
        }
    }
}

@Slf4j
@Component
@RequiredArgsConstructor
public class PlaceOrderHandler {

    private StackSagaKafkaTemplate<OrderDomainEntity, PlaceOrderTopic> stackSagaKafkaTemplate; (1)

    public String handle(String username, double amount, List<String> items) {
        String transactionId = this.stackSagaKafkaTemplate
                (2)
                .init(() -> {
                    OrderDomainEntity orderDomainEntity = new OrderDomainEntity();
                    {//initializing the domain entity with the required properties for the saga execution.
                        orderDomainEntity.setUsername(username);
                        orderDomainEntity.setTotalAmount(amount);
                        //...
                    }
                    return orderDomainEntity;
                })
                (3)
                .peek(orderDomainEntity -> {
                    log.info("transactionId {}:", orderDomainEntity.getTransactionId());
                })
                (4)
                .startWith(PlaceOrderTopic.DO_FETCH_USER_DETAILS, OrderEventManager.class)
                (5)
                .execute();
        log.info("Started place order saga with transactionId: {}", transactionId);
        return transactionId;
    }

    public void printCurrentState(String transactionId) {
        TransactionState<OrderDomainEntity> state = this
                .stackSagaKafkaTemplate
                (6)
                .getCurrentState(transactionId)
                .execute();

        TransactionCompleteStatus currentStatus = state.getCurrentStatus(); (7)
        log.info("Current status of transaction {}: {}", transactionId, currentStatus);
        NavigableMap<Integer, SagaExecutionEvent> executionHistory = state.getExecutionHistory();(8)
        log.info("Execution history of transaction {}: {}", transactionId, executionHistory);
        OrderDomainEntity currentDomainEntity = state.getCurrentDomainEntity(); (9)
        log.info("Current domain entity state: {}", currentDomainEntity); (10)
        ZonedDateTime startedDateTime = state.getStartedDateTime(); (11)
        log.info("Transaction {} started at: {}", transactionId, startedDateTime.toLocalDateTime());
    }
}

@Component
@RequiredArgsConstructor
class ReactivePlaceOrderHandler {
    private StackSagaKafkaTemplate<OrderDomainEntity, PlaceOrderTopic> stackSagaKafkaTemplate;

    public Mono<String> handle(String username, double amount, List<String> items) {
        return this
                .stackSagaKafkaTemplate
                .init(() -> {
                    OrderDomainEntity orderDomainEntity = new OrderDomainEntity();
                    {//initializing the domain entity with the required properties for the saga execution.
                        orderDomainEntity.setUsername(username);
                        orderDomainEntity.setTotalAmount(amount);
                        //...
                    }
                    return orderDomainEntity;
                })
                .peek(orderDomainEntity -> {
                    log.info("transactionId {}:", orderDomainEntity.getTransactionId());
                })
                .startWith(PlaceOrderTopic.DO_FETCH_USER_DETAILS, OrderEventManager.class)
                .executeAsync()
                .doOnNext(transactionId -> {
                    log.info("Started place order saga with transactionId: {}", transactionId);
                });
    }
    public Mono<Void> printAndCurrentState(String transactionId) {
        return this
                .stackSagaKafkaTemplate
                .getCurrentState(transactionId)
                .executeAsync()
                .doOnNext(state -> {
                    TransactionCompleteStatus currentStatus = state.getCurrentStatus();
                    log.info("Current status of transaction {}: {}", transactionId, currentStatus);
                    NavigableMap<Integer, SagaExecutionEvent> executionHistory = state.getExecutionHistory();
                    log.info("Execution history of transaction {}: {}", transactionId, executionHistory);
                    OrderDomainEntity currentDomainEntity = state.getCurrentDomainEntity();
                    log.info("Current domain entity state: {}", currentDomainEntity);
                    ZonedDateTime startedDateTime = state.getStartedDateTime();
                    log.info("Transaction {} started at: {}", transactionId, startedDateTime.toLocalDateTime());
                })
                .then();

    }
}

@Slf4j
@Component
@RequiredArgsConstructor
public class PlaceOrderHandler implements TransactionEventListener<OrderDomainEntity> {
    @Override
    public void onStateChanged(TransactionState<OrderDomainEntity> state) {
        TransactionCompleteStatus currentStatus = state.getCurrentStatus();
        log.info("Current status of transaction: {}", currentStatus);
        NavigableMap<Integer, SagaExecutionEvent> executionHistory = state.getExecutionHistory();
        log.info("Execution history of transaction: {}", executionHistory);
        OrderDomainEntity currentDomainEntity = state.getCurrentDomainEntity();
        log.info("Current domain entity state: {}", currentDomainEntity);
        ZonedDateTime startedDateTime = state.getStartedDateTime();
        log.info("Transaction started at: {}", startedDateTime.toLocalDateTime());
    }
}

@Slf4j
@Component
@RequiredArgsConstructor
public class ReactivePlaceOrderHandler implements ReactiveTransactionEventListener<OrderDomainEntity> {
    @Override
    public Mono<Void> onStateChanged(TransactionState<OrderDomainEntity> transactionState) {
        return Mono
                .just(transactionState)
                .doOnNext(state -> {
                    TransactionCompleteStatus currentStatus = state.getCurrentStatus();
                    log.info("Current status of transaction: {}", currentStatus);
                    NavigableMap<Integer, SagaExecutionEvent> executionHistory = state.getExecutionHistory();
                    log.info("Execution history of transaction: {}", executionHistory);
                    OrderDomainEntity currentDomainEntity = state.getCurrentDomainEntity();
                    log.info("Current domain entity state: {}", currentDomainEntity);
                    ZonedDateTime startedDateTime = state.getStartedDateTime();
                    log.info("Transaction started at: {}", startedDateTime.toLocalDateTime());
                })
                .flatMap(orderDomainEntityTransactionState -> {
                    //simulate some async processing here.
                    return Mono.delay(Duration.ofSeconds(1))
                            .then();
                });
    }
}

(2)
@SagaEndpoint(
        //the real topic name will be saga.do.user-service.validate-user
        topicNameSuffix = "user-service.get-user", (3)
        listenerScope = ListenerScope.SHARED (4)
)
public class UserValidateEndpoint extends QueryEndpoint { (1)

    (5)
    @Override
    public void doProcess(ConsumerRecord<String, SagaPayload> consumerRecord)
            throws JustRetryableExecutorException, RetryableExecutorException, NonRetryableExecutorException {

        log.info("Message Key (Transaction Id): {}", consumerRecord.key()); (6)
        log.info("IdempotencyKey: {}", consumerRecord.value().getIdempotencyKey()); (7)

        consumerRecord.value().getCurrentDomainEntityStateForUpdate().ifPresent(currentDomainEntityState -> { (8)
            log.info("Received payload for before user validation: {}", currentDomainEntityState);
            try {
                {
                    //user validation logic (9)
                    String username = currentDomainEntityState.get("username").asText();
                    log.info("Validating user with username: {}", username);
                }
                {(10)
                    // put the result into the domain entity state for the next topic to use.
                    currentDomainEntityState.put("is_user_validated", true);
                    currentDomainEntityState.put("validation_note", "user validated at " + LocalDateTime.now());
                }
            } catch (SomeRetryableException e) {
                (11)
                throw RetryableExecutorException.of(e);
            } catch (SomeNonRetryableException e) {
                (12)
                throw NonRetryableExecutorException
                        .buildWith(e)
                        .put("error_code", "USER_VALIDATION_FAILED")
                        .put("key-1", "value-1")
                        .put("key-2", "value-2")
                        .build();
            }
        });
    }
}

@Slf4j
@SagaEndpoint(
        topicNameSuffix = "payment.service.make-payment",
        listenerScope = ListenerScope.SHARED
)
public class MakePaymentEndpoint extends CommandEndpoint {

    @Override
    public void doProcess(ConsumerRecord<String, SagaPayload> consumerRecord)
            throws JustRetryableExecutorException, RetryableExecutorException, NonRetryableExecutorException {
        consumerRecord
                .value()
                .getCurrentDomainEntityStateForUpdate()
                .ifPresent(currentDomainEntityState -> {
                    final String paymentReferenceId = currentDomainEntityState.get("payment_reference_id").asText();
                    final double totalAmount = currentDomainEntityState.get("total_amount").asDouble();
                    try {
                        this.internalPaymentService.makePayment(paymentReferenceId, totalAmount);
                    } catch (SomeRetryableException e) {
                        RetryContext retryContext = consumerRecord.value().getRetryContext().orElseThrow();
                        if (retryContext.getRetryCount() >= 3) {
                            //if the retry count has reached the threshold, we can throw a NonRetryableExecutorException to stop further retries.
                            throw NonRetryableExecutorException
                                    .buildWith(e)
                                    .put("error_code", "PAYMENT_FAILED_AFTER_RETRIES")
                                    .build();
                        } else {
                            throw JustRetryableExecutorException.of("Payment failed, retrying... attempt " + (retryContext.getRetryCount() + 1));
                        }
                    } catch (SomeNonRetryableException e) {
                        throw NonRetryableExecutorException
                                .buildWith(e)
                                .put("error_code", "PAYMENT_FAILED")
                                .build();
                    }
                });
    }

    @Override
    public void undoProcess(ConsumerRecord<String, SagaPayload> consumerRecord) throws
            JustRetryableExecutorException, RetryableExecutorException {
    }
}

@Slf4j
@SagaEndpoint(
        //the real topic name will be saga.do.user-service.validate-user
        topicNameSuffix = "user-service.get-user",
        listenerScope = ListenerScope.SHARED
)
public class ReactiveUserValidateEndpoint extends ReactiveQueryEndpoint {(1)

    (2)
    @Override
    public Mono<Void> doProcess(ConsumerRecord<String, SagaPayload> consumerRecord) {
        log.info("Message Key (Transaction Id): {}", consumerRecord.key());
        log.info("IdempotencyKey: {}", consumerRecord.value().getIdempotencyKey());
        final ObjectNode currentDomainEntityState = consumerRecord
                .value()
                .getCurrentDomainEntityStateForUpdate()
                .orElseThrow();
        log.info("Received payload for before user validation: {}", currentDomainEntityState);
        String username = currentDomainEntityState.get("username").asText();
        return this.internalUserService
                .getUserDetails(username)
                .flatMap(userDetails -> {
                    {
                        //user validation logic
                    }
                    {
                        // put the result into the domain entity state for the next topic to use.
                        currentDomainEntityState.put("is_user_validated", true);
                        currentDomainEntityState.put("validation_note", "user validated at " + LocalDateTime.now());
                    }
                    return Mono.empty();
                })
                .onErrorResume(SomeRetryableException.class, e -> {
                    return Mono.error(RetryableExecutorException.of(e));
                })
                .onErrorResume(SomeNonRetryableException.class, e -> {
                    return Mono.error(NonRetryableExecutorException
                            .buildWith(e)
                            .put("error_code", "USER_VALIDATION_FAILED")
                            .build());
                });
    }
}

@Slf4j
@SagaEndpoint(
        topicNameSuffix = "payment.service.make-payment",
        listenerScope = ListenerScope.SHARED
)
class MakePaymentEndpoint extends CommandEndpoint { (1)

    (2)
    @Override
    public void doProcess(ConsumerRecord<String, SagaPayload> consumerRecord)
            throws JustRetryableExecutorException, RetryableExecutorException, NonRetryableExecutorException {
        //the theory is the same as the query endpoint.
    }

    (2)
    @Override
    public void undoProcess(ConsumerRecord<String, SagaPayload> consumerRecord) throws JustRetryableExecutorException, RetryableExecutorException {
        log.info("Message Key (Transaction Id): {}", consumerRecord.key());
        log.info("IdempotencyKey: {}", consumerRecord.value().getIdempotencyKey());
        (3)
        final JsonNode lastDomainEntityState = consumerRecord.value().getDomainEntityState();
        log.info("Last domain entity state {}", lastDomainEntityState);

        try {
            {//undo payment logic
                final double totalAmount = lastDomainEntityState.get("total_amount").asDouble();
                final String paymentRef = lastDomainEntityState.get("payment_reference_id").asText();
                this.internalPaymentSerivce.refundPayment(paymentRef, totalAmount);
                //update some hints for the next revert execution if needed.
                consumerRecord.value().getHintStore().ifPresent(hintStore -> {(4)
                    hintStore.put("last_refund_time", LocalDateTime.now().toString());
                });
            }
        } catch (SomeRetryableException e) {(5)
            throw RetryableExecutorException.of(e);
        }
    }
}

@Slf4j
@SagaEndpoint( (2)
        topicNameSuffix = "payment.service.make-payment",
        listenerScope = ListenerScope.SHARED
)
class ReactiveMakePaymentEndpoint extends ReactiveCommandEndpoint { (1)

    @Override
    public Mono<Void> doProcess(ConsumerRecord<String, SagaPayload> consumerRecord) {
        //the theory is the same as the query endpoint.
    }

    @Override
    public Mono<Void> undoProcess(ReceiverRecord<String, SagaPayload> consumerRecord) {
        log.info("Message Key (Transaction Id): {}", consumerRecord.key());
        log.info("IdempotencyKey: {}", consumerRecord.value().getIdempotencyKey());
        final JsonNode lastDomainEntityState = consumerRecord.value().getDomainEntityState();
        log.info("Last domain entity state {}", lastDomainEntityState);
        final double totalAmount = lastDomainEntityState.get("total_amount").asDouble();
        final String paymentRef = lastDomainEntityState.get("payment_reference_id").asText();
        this.internalPaymentSerivce
                .refundPayment(paymentRef, totalAmount)
                .flatMap(response -> {
                        //update some hints for the next revert execution if needed.
                        consumerRecord.value().getHintStore().ifPresent(hintStore -> {
                            hintStore.put("last_refund_time", LocalDateTime.now().toString());
                        });
                        return Mono.empty();
                })
                .onErrorResume(SomeRetryableException.class, e -> {(3)
                    return Mono.error(RetryableExecutorException.of(e));
                });
    }
}