fixing checkout consumer issue: still some issues

Author: jurabek

src/backend/services/checkout-api/src/index.ts

@@ -33,6 +33,7 @@ async function main() {
 
   process.on('SIGINT', async () => {
     await otel.shutdown();
+    await checkoutPublisher.shutdown();
     logger.info("Gracefully shutting down from SIGINT (Ctrl-C)");
     // some other closing procedures go here
     process.exit(0);

src/backend/services/checkout-api/src/messagging/publisher.ts

@@ -11,6 +11,7 @@ export class Publisher {
 
   public async start(): Promise<void> {
     try {
+      
       await this.producer.connect()
     } catch (error) {
       logger.error('Error connecting the producer: ', error)
@@ -32,6 +33,10 @@ export class Publisher {
     const kafka = new Kafka({
       clientId: this.clientId,
       brokers: this.brokers,
+      retry: {
+        initialRetryTime: 3000,
+        retries: 15
+      },
     })
     return kafka.producer()
   }

src/backend/services/order-api/src/main/java/org/jurabek/restaurant/order/api/events/UserCheckoutEventHandler.java

@@ -2,18 +2,15 @@
 
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
-
-import io.smallrye.common.annotation.Blocking;
-
-import java.util.concurrent.CompletionStage;
+import jakarta.transaction.Transactional;
 
 import org.eclipse.microprofile.faulttolerance.Retry;
-import org.eclipse.microprofile.reactive.messaging.Acknowledgment;
 import org.eclipse.microprofile.reactive.messaging.Incoming;
-import org.eclipse.microprofile.reactive.messaging.Message;
 import org.jboss.logging.Logger;
 import org.jurabek.restaurant.order.api.services.CheckoutService;
 
+import io.smallrye.common.annotation.Blocking;
+
 @ApplicationScoped
 public class UserCheckoutEventHandler {
 
@@ -27,17 +24,11 @@ public UserCheckoutEventHandler(CheckoutService checkout) {
     }
 
     @Incoming("checkout")
-    @Acknowledgment(Acknowledgment.Strategy.MANUAL)
-    @Retry(delay = 10, maxRetries = 5)
+    @Retry(maxRetries = 5)
+    @Transactional
     @Blocking
-    public CompletionStage<Void> Handle(Message<UserCheckoutEvent> message) {
-        try {
-            log.info("received user checkout event: " + message);
-            checkout.Checkout(message.getPayload());
-            return message.ack();
-        } catch (Exception e) {
-            log.error("Error processing user checkout event: " + message, e);
-            return message.nack(e);
-        }
+    public void Handle(UserCheckoutEvent message) {
+        log.info("received user checkout event: " + message);
+        checkout.Checkout(message);
     }
 }

src/backend/services/order-api/src/main/java/org/jurabek/restaurant/order/api/services/CheckoutService.java

@@ -1,10 +1,10 @@
 package org.jurabek.restaurant.order.api.services;
 
 import java.util.UUID;
+import java.util.concurrent.CompletionStage;
 
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
-import jakarta.transaction.Transactional;
 
 import org.eclipse.microprofile.reactive.messaging.Channel;
 import org.eclipse.microprofile.reactive.messaging.Emitter;
@@ -13,10 +13,9 @@
 import org.jurabek.restaurant.order.api.mappers.OrdersMapper;
 import org.jurabek.restaurant.order.api.repositories.OrdersRepository;
 
-
 @ApplicationScoped
 public class CheckoutService {
-   
+
     private final OrdersRepository ordersRepository;
     private final OrdersMapper mapper;
 
@@ -30,22 +29,19 @@ public CheckoutService(OrdersRepository ordersRepository, OrdersMapper mapper) {
         this.mapper = mapper;
     }
 
-    @Transactional
     public void Checkout(UserCheckoutEvent checkoutInfo) {
         var order = mapper.mapDtoToOrder(checkoutInfo.getCustomerBasket());
         order.setTransactionId(checkoutInfo.getTransactionId());
         order.setBuyerId(UUID.fromString(checkoutInfo.getCheckOutInfo().getUserId()));
         order.setCheckoutId(checkoutInfo.getCheckoutId());
-        
+
         for (var orderItems : order.getOrderItems()) {
             orderItems.setOrder(order);
         }
         ordersRepository.persist(order);
 
         var event = new OrderCompleted(order.getId(), order.getCartId(), order.getBuyerId(), order.getTransactionId(),
                 order.getOrderedDate());
-
-        orderCompletedEventEmitter.send(event);
+        orderCompletedEventEmitter.send(event).toCompletableFuture().join();
     }
-
 }