Microservices e-commerce #8: event-driven với Kafka

implementation 'org.springframework.cloud:spring-cloud-starter-stream-kafka'
testImplementation 'org.springframework.cloud:spring-cloud-stream-test-binder'

/**
 * Event dùng chung để tạo và xoá dữ liệu ở các core service.
 * key là productId, data là đối tượng cần tạo (null với DELETE).
 */
public record Event<K, T>(
    Type eventType,
    K key,
    T data,
    @JsonFormat(shape = JsonFormat.Shape.STRING) ZonedDateTime eventCreatedAt) {

    public enum Type {
        CREATE,
        DELETE
    }

    public Event(Type eventType, K key, T data) {
        this(eventType, key, data, ZonedDateTime.now());
    }
}

@Override
public Mono<Product> createProduct(Product body) {
    return Mono.fromCallable(() -> {
        sendMessage("products-out-0", new Event<>(Event.Type.CREATE, body.productId(), body));
        return body;
    }).subscribeOn(publishEventScheduler);
}

@Override
public Mono<Void> deleteProduct(int productId) {
    return Mono.fromRunnable(() -> sendMessage("products-out-0", new Event<>(Event.Type.DELETE, productId, null)))
        .subscribeOn(publishEventScheduler).then();
}

/** Gửi event qua StreamBridge; binding name quyết định topic (xem application.yml). */
private void sendMessage(String bindingName, Event<Integer, ?> event) {
    LOG.debug("Sending a {} message to {}", event.eventType(), bindingName);
    streamBridge.send(bindingName, event);
}

/**
 * StreamBridge.send() có thể bị chặn (chờ metadata từ Kafka), nên chạy nó trên thread pool riêng
 * thay vì event loop của WebFlux, giống cách review-service chạy JPA ở bài 6.
 */
@Bean
public Scheduler publishEventScheduler(
        @Value("${app.threadPoolSize:10}") Integer threadPoolSize,
        @Value("${app.taskQueueSize:100}") Integer taskQueueSize) {
    return Schedulers.newBoundedElastic(threadPoolSize, taskQueueSize, "publish-pool");
}

spring.cloud.stream:
  defaultBinder: kafka
  default.contentType: application/json
  bindings:
    products-out-0.destination: products
    recommendations-out-0.destination: recommendations
    reviews-out-0.destination: reviews
    inventory-out-0.destination: inventory
  kafka.binder.brokers: localhost:9092

@ResponseStatus(HttpStatus.ACCEPTED)
@PostMapping(value = "/product-composite", consumes = "application/json")
Mono<Void> createProduct(@RequestBody ProductAggregate body);

if (body.productId() < 1) {
    throw new InvalidInputException("Invalid productId: " + body.productId());
}

@Configuration
public class MessageProcessorConfig {

    private static final Logger LOG = LoggerFactory.getLogger(MessageProcessorConfig.class);

    private final ProductService productService;

    public MessageProcessorConfig(ProductService productService) {
        this.productService = productService;
    }

    /** Spring Cloud Stream gắn bean này vào binding messageProcessor-in-0, tức topic products. */
    @Bean
    public Consumer<Event<Integer, Product>> messageProcessor() {
        return event -> {
            LOG.info("Process message created at {}...", event.eventCreatedAt());

            switch (event.eventType()) {
                case CREATE -> {
                    LOG.info("Create product with ID: {}", event.key());
                    productService.createProduct(event.data()).block();
                }
                case DELETE -> {
                    LOG.info("Delete products with ProductID: {}", event.key());
                    productService.deleteProduct(event.key()).block();
                }
                default -> throw new EventProcessingException(
                    "Incorrect event type: " + event.eventType() + ", expected a CREATE or DELETE event");
            }

            LOG.info("Message processing done!");
        };
    }
}

spring.cloud.function.definition: messageProcessor

spring.cloud.stream:
  defaultBinder: kafka
  default.contentType: application/json
  bindings.messageProcessor-in-0:
    destination: products
    group: productsGroup
  kafka.binder.brokers: localhost:9092

  kafka:
    image: apache/kafka:3.9.1
    mem_limit: 1024m
    environment:
      # Chế độ KRaft: một node vừa là broker vừa là controller, không cần ZooKeeper
      - KAFKA_NODE_ID=1
      - KAFKA_PROCESS_ROLES=broker,controller
      - KAFKA_LISTENERS=PLAINTEXT://:9092,CONTROLLER://:9093
      - KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://kafka:9092
      - KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      - KAFKA_CONTROLLER_QUORUM_VOTERS=1@kafka:9093
      - KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1
      - KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1
      - KAFKA_TRANSACTION_STATE_LOG_MIN_ISR=1
      - KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0
    healthcheck:
      test: ["CMD-SHELL", "/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list"]
      interval: 10s
      timeout: 10s
      retries: 30

    depends_on:
      kafka:
        condition: service_healthy

  assertCurl 202 "curl -X DELETE http://$HOST:$PORT/product-composite/${productId} -s"
  assertEqual 202 $(curl -X POST -s http://$HOST:$PORT/product-composite -H "Content-Type: application/json" \
    --data "$composite" -w "%{http_code}")

# Ghi giờ là bất đồng bộ: chờ tới khi các service lõi đã xử lý xong event của dữ liệu test
function waitForMessageProcessing() {
  echo "Wait for messages to be processed... "
  local n=0
  until [ "$(curl -s http://$HOST:$PORT/product-composite/$PROD_ID_REVS_RECS \
      | jq -c '[(.recommendations | length), (.reviews | length), .stock]' 2>/dev/null)" = "[3,3,10]" ] \
    && [ "$(curl -s -o /dev/null -w "%{http_code}" http://$HOST:$PORT/product-composite/$PROD_ID_NO_RECS)" = "200" ] \
    && [ "$(curl -s -o /dev/null -w "%{http_code}" http://$HOST:$PORT/product-composite/$PROD_ID_NO_REVS)" = "200" ]; do
    n=$((n + 1))
    if [[ $n == 40 ]]; then
      echo " Give up"
      exit 1
    fi
    sleep 1
    echo -n ", retry #$n "
  done
  echo "All messages are now processed!"
}

# Xoá là idempotent: gửi hai lần đều được nhận (202), sau khi event được xử lý thì sản phẩm không còn
assertCurl 202 "curl -X DELETE http://$HOST:$PORT/product-composite/$PROD_ID_NO_REVS -s"
assertCurl 202 "curl -X DELETE http://$HOST:$PORT/product-composite/$PROD_ID_NO_REVS -s"
waitForHttpCode 404 http://$HOST:$PORT/product-composite/$PROD_ID_NO_REVS
assertCurl 404 "curl http://$HOST:$PORT/product-composite/$PROD_ID_NO_REVS -s"

$ ./gradlew build && docker compose build
$ ./test-em-all.bash start
...
Wait for: http://localhost:8080/product-composite/13... , retry #1 , ... , retry #7 DONE, continues...
...
Wait for messages to be processed...
, retry #1 All messages are now processed!
...
End, all tests OK: Thu Oct 8 03:36:35 UTC 2026

$ curl -s -i -X POST localhost:8080/product-composite -H "Content-Type: application/json" \
    --data '{"productId":2,"name":"Balo laptop","price":650000,"stock":5,
             "reviews":[{"reviewId":1,"author":"an","subject":"Ben","content":"Dung tot"}]}' | head -1
HTTP/1.1 202 Accepted

$ docker compose exec -T kafka /opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
__consumer_offsets
inventory
products
recommendations
reviews

$ docker compose exec -T kafka /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
    --topic products --from-beginning --timeout-ms 5000
...
{"eventType":"DELETE","key":213,"data":null,"eventCreatedAt":"2026-10-08T03:36:34.139536497Z"}
{"eventType":"DELETE","key":213,"data":null,"eventCreatedAt":"2026-10-08T03:36:34.17359037Z"}
{"eventType":"CREATE","key":2,"data":{"productId":2,"name":"Balo laptop","price":650000,"serviceAddress":null},"eventCreatedAt":"2026-10-08T03:36:46.254659284Z"}

$ docker compose exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --describe --group productsGroup
GROUP           TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG ...
productsGroup   products        0          9               9               0   ...

INFO 1 --- [product] [container-0-C-1] c.e.c.p.services.MessageProcessorConfig  : Process message created at 2026-10-08T03:36:46.254659284Z...
INFO 1 --- [product] [container-0-C-1] c.e.c.p.services.MessageProcessorConfig  : Create product with ID: 2
INFO 1 --- [product] [container-0-C-1] c.e.c.p.services.MessageProcessorConfig  : Message processing done!

$ docker compose stop review
$ curl -s -o /dev/null -w "POST %{http_code}\n" -X POST localhost:8080/product-composite \
    -H "Content-Type: application/json" \
    --data '{"productId":3,"name":"Mu len","price":150000,
             "reviews":[{"reviewId":1,"author":"binh","subject":"Am","content":"Doi mua dong rat am"}]}'
POST 202
$ curl -s localhost:8080/product-composite/3 | jq -c '{productId,name,reviews: (.reviews|length)}'
{"productId":3,"name":"Mu len","reviews":0}

$ docker compose exec -T kafka /opt/kafka/bin/kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
    --describe --group reviewsGroup
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID
reviewsGroup reviews 0 12 13 1 -

$ docker compose start review
$ curl -s localhost:8080/product-composite/3 | jq -c '{productId,name,reviews}'
{"productId":3,"name":"Mu len","reviews":[{"reviewId":1,"author":"binh","subject":"Am","content":"Doi mua dong rat am"}]}

INFO 1 --- [review] [container-0-C-1] c.e.c.r.services.MessageProcessorConfig  : Process message created at 2026-10-08T03:37:12.421690509Z...
INFO 1 --- [review] [container-0-C-1] c.e.c.r.services.MessageProcessorConfig  : Create review with ID: 3
INFO 1 --- [review] [container-0-C-1] c.e.c.r.services.MessageProcessorConfig  : Message processing done!

$ curl -s -o /dev/null -w "POST %{http_code}\n" -X POST localhost:8080/product-composite \
    -H "Content-Type: application/json" --data '{"productId":1,"name":"Giay sneaker","price":899000}'
POST 202

ERROR 1 --- [product] [container-0-C-1] o.s.integration.handler.LoggingHandler   : org.springframework.messaging.MessageHandlingException: error occurred in message handler [...], failedMessage=GenericMessage [payload=byte[162], headers={kafka_offset=10, ..., deliveryAttempt=3, ..., kafka_receivedTopic=products, ..., kafka_groupId=productsGroup}]
Caused by: com.ecommerce.api.exceptions.InvalidInputException: Duplicate key, Product Id: 1
ERROR 1 --- [product] [container-0-C-1] o.s.kafka.listener.DefaultErrorHandler   : Backoff none exhausted for products-0@10

git add .
git commit -m "Bài 8: event-driven với Kafka"
git tag blog-08