stacksaga-kafka-worker — Endpoints

(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(
        //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(
        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 {
    }
}

@Configuration
public class EndpointSchedulerConfig {

    @Bean
    public AbstractDefaultPrimaryExecutionSchedulerProvider customSharedPrimaryExecutionSchedulerProvider() {
        return new AbstractDefaultPrimaryExecutionSchedulerProvider() {
            @Override
            protected Scheduler scheduler() {
                log.info("lazy initialization of custom shared primary execution scheduler for non-reactive(blocking) execution");
                return Schedulers.newBoundedElastic(
                        100,
                        Integer.MAX_VALUE,
                        "custom-do-shared",
                        60,
                        false
                );
            }
        };
    }

    @Bean
    public AbstractDefaultRevertExecutionSchedulerProvider customSharedRevertExecutionSchedulerProvider() {
        return new AbstractDefaultRevertExecutionSchedulerProvider() {
            @Override
            protected Scheduler scheduler() {
                log.info("lazy initialization of custom shared revert execution scheduler for non-reactive(blocking) execution");
                return Schedulers.newBoundedElastic(
                        20,
                        Integer.MAX_VALUE,
                        "custom-undo-shared",
                        60,
                        false
                );
            }
        };
    }
}

@Component
public class MakePaymentExecutionSchedulerProvider extends AbstractSagaEndpointSchedulerProvider {
    @Override
    protected Scheduler scheduler() {
        return Schedulers.newBoundedElastic(20, Integer.MAX_VALUE, "demo-name");
    }
}


@Slf4j
@SagaEndpoint(
        listenerScope = ListenerScope.SHARED,
        primaryExecutionSchedulerProvider = MakePaymentExecutionSchedulerProvider.class,
        revertExecutionSchedulerProvider = MakePaymentExecutionSchedulerProvider.class
)
public class MakePaymentEndpoint extends CommandEndpoint {
    //...
}

@Component
public class MakePaymentRetryTemplateProvider extends AbstractRetryTemplateProvider {
    @Override
    protected RetryTemplateBuilder retryTemplateBuilder() {
        SimpleRetryPolicy simpleRetryPolicy = new SimpleRetryPolicy();
        simpleRetryPolicy.setMaxAttempts(3);
        return RetryTemplate.builder().customPolicy(simpleRetryPolicy);
    }
}

@Slf4j
@SagaEndpoint(
        topicNameSuffix = "payment.service.make-payment",
        listenerScope = ListenerScope.SHARED,
        primaryExecutionRetryTemplate = MakePaymentRetryTemplateProvider.class, (1)
        revertExecutionRetryTemplate = MakePaymentRetryTemplateProvider.class (2)
)
public class MakePaymentEndpoint extends CommandEndpoint {
    //...
}

@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));
                });
    }
}