I am following a tutorial until timestamp: 05:54:05
🚀 🔥 Mastering Micro-services: Spring boot, Spring Cloud and Keycloak In 7 Hours -> Timestamp: 05:54:05

Here is the link for the project Github Repo

and got the following exception:

2024-09-10T16:10:49.762-07:00 ERROR 92288 --- Backoff FixedBackOff{interval=0, currentAttempts=1, maxAttempts=0} exhausted for payment-topic-0@5

Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.microservices.ecommerce.kafka.NotificationConsumer.consumePaymentSuccessNotification(com.microservices.ecommerce.kafka.payment.PaymentConfirmation) throws jakarta.mail.MessagingException]
Bean [com.microservices.ecommerce.kafka.NotificationConsumer@6615ccb6]
    

Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot handle message
    
Caused by: org.springframework.messaging.converter.MessageConversionException:  Cannot convert from [java.lang.String] to [com.microservices.ecommerce.kafka.payment.PaymentConfirmation] for GenericMessage [payload={"orderReference":"MS-20241201","amount":900,"paymentMethod":"PAYPAL","customerFirstname":null,"customerLastname":null,"customerEmail":"[email protected]"}, headers={kafka_offset=5, kafka_consumer=org.springframework.kafka.core.DefaultKafkaConsumerFactory$ExtendedKafkaConsumer@3543a9a7, kafka_timestampType=CREATE_TIME, kafka_receivedPartitionId=0, kafka_receivedTopic=payment-topic, kafka_receivedTimestamp=1726009849728, __TypeId__=[B@16b8789b, kafka_groupId=paymentGroup}]

2024-09-10T16:10:49.762-07:00 ERROR 92288 --- Backoff FixedBackOff{interval=0, currentAttempts=1, maxAttempts=0} exhausted for order-topic-0@5

ListenerExecutionFailedException: Listener method could not be invoked with the incoming message
Endpoint handler details:
Method [public void com.microservices.ecommerce.kafka.NotificationConsumer.consumeOrderConfirmationNotification(com.microservices.ecommerce.kafka.order.OrderConfirmation) throws jakarta.mail.MessagingException]
Bean [com.microservices.ecommerce.kafka.NotificationConsumer@6615ccb6]

Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot handle message

Caused by: org.springframework.messaging.converter.MessageConversionException: Cannot convert from [java.lang.String] to [com.microservices.ecommerce.kafka.order.OrderConfirmation] for GenericMessage [payload={"orderReference":"MS-20241201","totalAmount":900,"paymentMethod":"PAYPAL","customerResponse":{"id":"66e09b55f7786369d16fdfff","firstname":null,"lastname":null,"email":"[email protected]"},"products":[{"productId":1,"name":"Mechanical Keyboard 1","description":"Mechanical keyboard with RGB lighting","price":99.99,"quantity":1.0},{"productId":251,"name":"4K Monitor 1","description":"27-inch IPS monitor with 4K resolution","price":399.99,"quantity":1.0},{"productId":851,"name":"Ambidextrous Gaming Mouse 1","description":"Ambidextrous gaming mouse with high DPI","price":69.99,"quantity":1.0}]}, headers={kafka_offset=5, kafka_consumer=org.springframework.kafka.core.DefaultKafkaConsumerFactory$ExtendedKafkaConsumer@33222867, kafka_timestampType=CREATE_TIME, kafka_receivedPartitionId=0, kafka_receivedTopic=order-topic, kafka_receivedTimestamp=1726009849737, __TypeId__=[B@7436f23e, kafka_groupId=orderGroup}]

Long story short, this micro-service project structure:
[ Producer: order-service & payment-service –> Consumer: notification-service ]

I think the problem may coming from consumer due to producer had successfully sent the confirmation:

2024-09-10T16:10:49.732-07:00  INFO 47674 --- Sending order confirmation: OrderConfirmation[orderReference=MS-20241201, totalAmount=900, paymentMethod=PAYPAL, customerResponse=CustomerResponse[id=xxxx, xxx=null, xxx=null, [email protected]], products=[PurchaseResponse[productId=1, name=Mechanical Keyboard 1, description=Mechanical keyboard with RGB lighting, price=99.99, quantity=1.0], PurchaseResponse[productId=251, name=4K Monitor 1, description=27-inch IPS monitor with 4K resolution, price=399.99, quantity=1.0], PurchaseResponse[productId=851, name=Ambidextrous Gaming Mouse 1, description=Ambidextrous gaming mouse with high DPI, price=69.99, quantity=1.0]]]

Hence the problem should coming from consumer unable to transfer String to Json format:

Cannot convert from [java.lang.String] to [] for GenericMessage [xxx]

order-service.yml

spring:
  datasource:
    url: jdbc:postgresql://localhost:5432/order
    username: root
    password: 123456
    driver-class-name: org.postgresql.Driver
  jpa:
    hibernate:
      ddl-auto: create
    database: postgresql
    database-platform: org.hibernate.dialect.PostgreSQLDialect
  kafka:
    producer:
      bootstrap-servers: "localhost:9092"
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        spring.json.type.mapping: orderConfirmation:com.microservices.ecommerce.kafka.OrderConfirmation

server:
  port: 8070

application:
  config:
    customer-url: http://localhost:8222/api/v1/customers
    payment-url: http://localhost:8222/api/v1/payments
    product-url: http://localhost:8222/api/v1/products

payment-service.yml

spring:
  datasource:
    url: jdbc:postgresql://localhost:5432/payment
    username: root
    password: 123456
    driver-class-name: org.postgresql.Driver
  jpa:
    hibernate:
      ddl-auto: update
    database: postgresql
    database-platform: org.hibernate.dialect.PostgreSQLDialect

  kafka:
    producer:
      bootstrap-servers: "localhost:9092"
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      properties:
        spring.json.type.mapping: paymentConfirmation:com.microservices.ecommerce.notification.PaymentNotificationRequest

server:
  port: 8060

notification-service.yml

spring:
  kafka:
    consumer:
      bootstrap-servers: "localhost:9092"
      group-id: paymentGroup,orderGroup
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.trusted.packages: '*'
        spring.json.type.mapping: orderConfirmation:com.microservices.ecommerce.kafka.order.OrderConfirmation,paymentConfirmation:com.microservices.ecommerce.kafka.payment.PaymentConfirmation
  mail:
    host: localhost
    port: 1025
    username: root
    password: 123456

    properties:
      mail:
        smtp:
          ssl:
            trust: "*"
          auth: true
          starttls:
            enable: true
            required: true
          connectiontimeout: 5000
          timeout: 3000
          writetimeout: 5000

server:
  port: 8040

notificationConsumer.java

@Service
@RequiredArgsConstructor
@Slf4j
public class NotificationConsumer {

    private final NotificationRepository repository;
    private final EmailService emailService;
    
    @KafkaListener(topics = "payment-topic", groupId = "paymentGroup", containerFactory = "kafkaListenerContainerFactory")
    public void consumePaymentSuccessNotification(PaymentConfirmation paymentConfirmation) throws MessagingException {
        log.info("Consuming the message from payment-topic Topic:: {}", paymentConfirmation.toString());
        repository.save(
                Notification.builder()
                        .type(PAYMENT_CONFIRMATION)
                        .notificationDate(LocalDateTime.now())
                        .paymentConfirmation(paymentConfirmation)
                        .build()
        );

        var customerName = paymentConfirmation.customerFirstname() + " " + paymentConfirmation.customerLastname();
        emailService.sendPaymentSuccessEmail(
                paymentConfirmation.customerEmail(),
                customerName,
                paymentConfirmation.amount(),
                paymentConfirmation.orderReference()
        );
    }

    @KafkaListener(topics = "order-topic", groupId = "orderGroup", containerFactory = "kafkaListenerContainerFactory")
    public void consumeOrderConfirmationNotification(OrderConfirmation orderConfirmation) throws MessagingException {
        log.info("Consuming the message from order-topic Topic:: {}", orderConfirmation.toString());
        repository.save(
                Notification.builder()
                        .type(ORDER_CONFIRMATION)
                        .notificationDate(LocalDateTime.now())
                        .orderConfirmation(orderConfirmation)
                        .build()
        );

        // TODO send email
        var customerName = orderConfirmation.customer().firstname() + " " + orderConfirmation.customer().lastname();
        emailService.sendOrderConfirmationEmail(
                orderConfirmation.customer().email(),
                customerName,
                orderConfirmation.totalAmount(),
                orderConfirmation.orderReference(),
                orderConfirmation.products()
        );
    }

}

config.yml

@Configuration
public class Config {

    @Bean
    public JavaMailSender javaMailSender() {
        return new JavaMailSenderImpl();
    }

}

Not sure I may need additional setting for consumer or Bean

1. Already checked the

spring.kafka.consumer.properties.spring.json.type.mapping 

and compare both class to make sure there is no attribute mismatch

2. Already checkout producer and consumer’s application.yml key and value both serializer and deserializer setting