快速入门示例：StackSaga-Kafka:MySQL

<dependencyManagement> (1)
    <dependencies>
        <dependency>
            <groupId>org.stacksaga</groupId>
            <artifactId>stacksaga-bom</artifactId>
            <version>1.0.0-SNAPSHOT</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>
<dependencies>
    <!-- Spring Boot Starter Web -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>
    <dependency>
      <groupId>org.springframework.boot</groupId>
      <artifactId>spring-boot-starter-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.projectlombok</groupId>
        <artifactId>lombok</artifactId>
        <optional>true</optional>
    </dependency>
    <dependency> (2)
        <groupId>org.stacksaga</groupId>
        <artifactId>stacksaga-kafka-orchestrator-spring-boot-starter</artifactId>
    </dependency>
    <dependency> (3)
        <groupId>org.stacksaga</groupId>
        <artifactId>stacksaga-mysql-reactive-support</artifactId>
    </dependency>
</dependencies>
<build>
    <plugins>
        <plugin> (4)
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-maven-plugin</artifactId>
            <executions>
                <execution>
                    <goals>
                        <goal>build-info</goal>
                    </goals>
                </execution>
            </executions>
        </plugin>
    </plugins>
</build>

@Getter
@Setter
@SagaDomainEntity(
        version = @SagaDomainEntityVersion(major = 1, minor = 0, patch = 0),
        name = "OrderDomainEntity"
)(1)
public class OrderDomainEntity extends DomainEntity { (2)

    // Domain-specific fields
    @JsonProperty("username")
    private String username;
    @JsonProperty("user_validated")
    private boolean userValidated;
    @JsonProperty("total_amount")
    private double totalAmount;
    @JsonProperty("payment_reference_id")
    private String paymentReferenceId;
    @JsonProperty("product_items")
    private List<String> productItems;

    (3)
    public OrderDomainEntity() {
        super(OrderDomainEntity.class);
    }
}

public class PlaceOrderTopic extends AbstractTopic<PlaceOrderTopic> { (1)

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


     (3)
    //do topics
    public static final PlaceOrderTopic DO_USER_VALIDATED = new PlaceOrderTopic("user-service.validate-user", 1.0f, SagaEventType.QUERY_DO_ACTION, "user-service");
    public static final PlaceOrderTopic DO_ORDER_RESERVED = new PlaceOrderTopic("order-service.reserve-order", 2.0f, SagaEventType.COMMAND_DO_ACTION, "order-service");
    public static final PlaceOrderTopic DO_MAKE_PAYMENT = new PlaceOrderTopic("payment-service.make-payment", 3.0f, SagaEventType.COMMAND_DO_ACTION, "payment-service");

    //undo topics
    public static final PlaceOrderTopic UNDO_ORDER_RESERVED = new PlaceOrderTopic("order-service.reserve-order", -2.0f, SagaEventType.COMMAND_UNDO_ACTION, "order-service", DO_ORDER_RESERVED);
    public static final PlaceOrderTopic UNDO_MAKE_PAYMENT = new PlaceOrderTopic("payment-service.make-payment", -3.0f, SagaEventType.COMMAND_UNDO_ACTION, "payment-service", DO_MAKE_PAYMENT);
}

(2)
@SagaEventManager(
        value = "placeOrderEventManager",
        listenerScope = OrchestratorListenerScope.SHARED_GLOBAL,
        groupType = GroupType.SHARE,
        domainCallbackTopicSuffix = "place-order"
)
public class PlaceOrderEventManager extends AbstractEventManager<OrderDomainEntity, PlaceOrderTopic> { (1)

    (3)
    @Override
    public Supplier<List<PlaceOrderTopic>> registerTopics() {
        return () -> List.of(
                PlaceOrderTopic.DO_USER_VALIDATED,
                PlaceOrderTopic.DO_ORDER_RESERVED,
                PlaceOrderTopic.UNDO_ORDER_RESERVED,
                PlaceOrderTopic.DO_MAKE_PAYMENT,
                PlaceOrderTopic.UNDO_MAKE_PAYMENT
        );
    }

    @NonNull
    @Override
    (4)
    public SagaPrimaryEventAction<PlaceOrderTopic> onNext(
            PlaceOrderTopic recentTopic,
            OrderDomainEntity currentDomainEntityState,
            SagaPrimaryEventActionUtil<PlaceOrderTopic> actionUtil
    ) {
        (5)
        if (recentTopic.equals(PlaceOrderTopic.DO_USER_VALIDATED)) {
            return actionUtil.next(PlaceOrderTopic.DO_ORDER_RESERVED);
        }
        (6)
        if (recentTopic.equals(PlaceOrderTopic.DO_ORDER_RESERVED)) {
            return actionUtil.next(PlaceOrderTopic.DO_MAKE_PAYMENT);
        }
        (7)
        if (recentTopic.equals(PlaceOrderTopic.DO_MAKE_PAYMENT)) {
            return actionUtil.complete();
        }
        (8)
        return actionUtil.error(new IllegalStateException("Unexpected topic: " + recentTopic));
    }
}

@Slf4log
@Component(1)
public class PlaceOrderHandler implements KafkaTransactionEventListener<OrderDomainEntity> {(2)
    @Override(3)
    public void onStateChanged(TransactionState<OrderDomainEntity, AsyncExecutionEvent> transactionState) {
        log.info("onStateChanged : {}", transactionState.getCurrentStatus());
    }
}

@Slf4j
@RestController
@RequestMapping("/api/v1/order")
@RequiredArgsConstructor
public class PlaceOrderController {

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

    @PostMapping
    public String placeOrder(@RequestBody PlaceOrderRequest placeOrderRequest) {
        final String transactionId = this
                .stackSagaKafkaTemplate
                (2)
                .init(() -> {
                    OrderDomainEntity orderDomainEntity = new OrderDomainEntity();
                    orderDomainEntity.setUsername(placeOrderRequest.getUsername());
                    orderDomainEntity.setTotalAmount(placeOrderRequest.getTotalAmount());
                    orderDomainEntity.setProductItems(Arrays.asList(placeOrderRequest.getItems()));
                    return orderDomainEntity;
                })
                (3)
                .peek(orderDomainEntity -> {
                    //you can do some local operation for the orderDomainEntity before the saga process.
                    //and also here you can access the unique id for the transaction that provide by the engine.
                    log.info("Transaction Id : {}", orderDomainEntity.getTransactionId());
                })
                (4)
                .startWith(PlaceOrderTopic.DO_USER_VALIDATED, PlaceOrderEventManager.class)
                (5)
                .execute();
        (6)
        return "Order placed successfully with transaction id: " + transactionId;
    }

    (7)
    @Data
    public static class PlaceOrderRequest {
        private String username;
        private double totalAmount;
        private String[] items;
    }
}

#spring properties
spring.application.name=order-service
#stacksaga-instance properties (1)
stacksaga.instance.cluster=local-cluster
stacksaga.instance.region=local-region
stacksaga.instance.zone=local-zone
#stacksaga-starter properties
stacksaga.kafka.orchestrator.domain-entity-scan=org.example.orderservice.domain (2)
#stacksaga-mysql-database-support properties (3)
stacksaga.mysql.r2dbc.url=r2dbc:mysql://localhost:3306
stacksaga.mysql.r2dbc.database=order_service_event_store
stacksaga.mysql.r2dbc.username=${MYSQL_USER:root}
stacksaga.mysql.r2dbc.password=${MYSQL_PASSWORD:password}
(4)
stacksaga.mysql.jdbc.url=jdbc:mysql://localhost:3306/order_service_event_store
stacksaga.mysql.jdbc.username=${MYSQL_USER:root}
stacksaga.mysql.jdbc.password=${MYSQL_PASSWORD:password}
(5)
spring.kafka.bootstrap-servers=${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.stacksaga</groupId>
            <artifactId>stacksaga-bom</artifactId>
            <version>1.0.0-SNAPSHOT</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>
<dependency>
    <groupId>org.stacksaga</groupId>
    <artifactId>stacksaga-kafka-worker-spring-boot-starter</artifactId>
</dependency>

@Slf4j
@SagaEndpoint(topicNameSuffix = "user-service.validate-user", listenerScope = WorkerListenerScope.SHARED_GLOBAL, groupType = GroupType.SHARE) (2)
public class UserValidationEndpoint extends QueryEndpoint { (1)

    @Override (3)
    public void doProcess(ConsumerRecord<String, SagaPayload> consumerRecord)
            throws JustRetryableExecutorException, RetryableExecutorException, NonRetryableExecutorException {
        log.info("Transaction Id: {},Idempotency Key: {}", consumerRecord.value().getTransactionId(), consumerRecord.value().getIdempotencyKey());
        consumerRecord.value().getCurrentDomainEntityStateForUpdate().ifPresent(domainEntity -> {
            String username = domainEntity.get("username").asText();
            log.info("username: {}", username);
            //Validate username with your logic here and if the user is not valid you can throw NonRetryableExecutorException
            //If there is a transient error, you can throw RetryableExecutorException or JustRetryableExecutorException

    try {
                //simulate some delay for make the request
                Thread.sleep(new Random().nextInt(1000, 3000));
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }

        });
    }
}

spring.application.name=user-service
#kafka connection (1)
spring.kafka.bootstrap-servers=${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
#stacksaga-instance properties (2)
stacksaga.instance.cluster=local-cluster
stacksaga.instance.region=local-region
stacksaga.instance.zone=local-zone

@Slf4j
@SagaEndpoint(topicNameSuffix = "order-service.reserve-order", listenerScope = WorkerListenerScope.SHARED_GLOBAL, groupType = GroupType.SHARE) (2)
public class ReserverOrderEndpoint extends CommandEndpoint { (1)

    (3)
    @Override
    public void doProcess(ConsumerRecord<String, SagaPayload> consumerRecord)
            throws JustRetryableExecutorException, RetryableExecutorException, NonRetryableExecutorException {
        log.info("Transaction Id: {},Idempotency Key: {}", consumerRecord.value().getTransactionId(), consumerRecord.value().getIdempotencyKey());
        consumerRecord.value().getCurrentDomainEntityStateForUpdate().ifPresent(domainEntity -> {
            String username = domainEntity.get("username").asText();
            log.info("username: {}", username);
            JsonNode productItems = domainEntity.get("product_items");
            for (JsonNode productItem : productItems) {
                String productId = productItem.asText();
                log.info("productId: {} ", productId);
            }
            try {
                //simulate some delay for make the request
                Thread.sleep(new Random().nextInt(1000, 3000));
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            // TODO: Reserve order with your logic here
        });
    }

    (4)
    @Override
    public void undoProcess(ConsumerRecord<String, SagaPayload> consumerRecord)
            throws JustRetryableExecutorException, RetryableExecutorException {
        SagaPayload payload = consumerRecord.value();
        log.info("Transaction Id: {},Idempotency Key: {}", payload.getTransactionId(), payload.getIdempotencyKey());
        JsonNode lastDomainEntityState = payload.getDomainEntityState();
        String username = lastDomainEntityState.get("username").asText();
        log.info("username: {}", username);
        payload.getPrimaryExecutionException().ifPresent(primaryExecutionExceptionMetaData -> {
            log.info("primary error message : {}", primaryExecutionExceptionMetaData.getRealExceptionMessage());
        });
        // TODO: revert the reserve order with your logic here
            try {
                //simulate some delay for make the request
                Thread.sleep(new Random().nextInt(1000, 3000));
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        (5)
        payload.getHintStore().ifPresent(hintStore -> {
            hintStore.put("undone_reserve_order", "true");
        });
    }
}

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.stacksaga</groupId>
            <artifactId>stacksaga-bom</artifactId>
            <version>1.0.0-SNAPSHOT</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>
<dependency>
    <groupId>org.stacksaga</groupId>
    <artifactId>stacksaga-kafka-worker-spring-boot-starter</artifactId>
</dependency>

@Slf4log
@SagaEndpoint(topicNameSuffix = "payment-service.make-payment", listenerScope = WorkerListenerScope.SHARED_GLOBAL, groupType = GroupType.SHARE)
public class MakePaymentEndpoint extends CommandEndpoint {
    @Override
    public void doProcess(ConsumerRecord<String, SagaPayload> consumerRecord)
            throws JustRetryableExecutorException, RetryableExecutorException, NonRetryableExecutorException {
        SagaPayload payload = consumerRecord.value();
        log.info("Transaction Id: {},Idempotency Key: {}", payload.getTransactionId(), payload.getIdempotencyKey());
        payload.getCurrentDomainEntityStateForUpdate().ifPresent(domainEntity -> {
            String username = domainEntity.get("username").asText();
            log.info("username: {}", username);
            double totalAmount = domainEntity.get("total_amount").asDouble();
            log.info("total amount: {}", totalAmount);
            // TODO: Make payment with your logic here
            try {
                //simulate some delay for make the request
                Thread.sleep(new Random().nextInt(1000, 3000));
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            //simulate a payment failure to demonstrate compensation.
            //run the request a few times until this branch is hit: the orchestrator will then
            //compensate the previously completed command span (UNDO_ORDER_RESERVED).
            if (new Random().nextBoolean()) {
                throw NonRetryableExecutorException
                        .buildWith(new RuntimeException("insufficient balance"))
                        .put("error_code", "MAKE_PAYMENT_FAILED")
                        .build();
            }
        });
    }

    @Override
    public void undoProcess(ConsumerRecord<String, SagaPayload> consumerRecord)
            throws JustRetryableExecutorException, RetryableExecutorException {
        SagaPayload payload = consumerRecord.value();
        log.info("Transaction Id: {},Idempotency Key: {}", payload.getTransactionId(), payload.getIdempotencyKey());
        JsonNode lastDomainEntityState = payload.getDomainEntityState();
        String username = lastDomainEntityState.get("username").asText();
        log.info("username: {}", username);
        double totalAmount = lastDomainEntityState.get("total_amount").asDouble();
        log.info("total amount: {}", totalAmount);

        payload.getPrimaryExecutionException().ifPresent(primaryExecutionExceptionMetaData -> {
            log.info("primary error message : {}", primaryExecutionExceptionMetaData.getRealExceptionMessage());
        });
        // TODO: revert the payment with your logic here
        payload.getHintStore().ifPresent(hintStore -> {
            hintStore.put("undone_make_payment", "true");
        });
    }
}

spring.application.name=payment-service
#kafka connection (1)
spring.kafka.bootstrap-servers=${KAFKA_BOOTSTRAP_SERVERS:localhost:9092}
#stacksaga-instance properties (2)
stacksaga.instance.cluster=local-cluster
stacksaga.instance.region=local-region
stacksaga.instance.zone=local-zone