fix(consumer): harden job event listeners with idempotency, DLQ, and …

src/main/java/com/isums/maintainservice/configurations/KafkaConsumerConfig.java

@@ -0,0 +1,70 @@
+package com.isums.maintainservice.configurations;
+
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.kafka.core.DefaultKafkaProducerFactory;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
+import org.springframework.kafka.listener.DefaultErrorHandler;
+import org.springframework.util.backoff.ExponentialBackOff;
+
+import java.util.Map;
+
+@Configuration
+public class KafkaConsumerConfig {
+
+    @Value("${spring.kafka.bootstrap-servers}")
+    private String bootstrapServers;
+
+    @Bean
+    public KafkaTemplate<String, Object> objectKafkaTemplate() {
+        Map<String, Object> props = Map.of(
+                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers,
+                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
+                org.springframework.kafka.support.serializer.JsonSerializer.class,
+                org.springframework.kafka.support.serializer.JsonSerializer.ADD_TYPE_INFO_HEADERS, false
+        );
+        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
+    }
+
+    @Bean
+    public KafkaTemplate<String, String> dltKafkaTemplate() {
+        Map<String, Object> props = Map.of(
+                ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers,
+                ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class,
+                ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class
+        );
+        return new KafkaTemplate<>(new DefaultKafkaProducerFactory<>(props));
+    }
+
+    @Bean
+    public DefaultErrorHandler kafkaErrorHandler(KafkaTemplate<String, String> dltKafkaTemplate) {
+        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
+                dltKafkaTemplate,
+                (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition())
+        );
+
+        ExponentialBackOff backOff = new ExponentialBackOff(1_000L, 2.0);
+        backOff.setMaxInterval(60_000L);
+        backOff.setMaxAttempts(Long.MAX_VALUE);
+
+        DefaultErrorHandler handler = new DefaultErrorHandler(recoverer, backOff);
+
+        handler.addNotRetryableExceptions(
+                com.fasterxml.jackson.core.JsonProcessingException.class,
+                com.fasterxml.jackson.databind.exc.InvalidDefinitionException.class,
+                com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException.class,
+                IllegalArgumentException.class,
+                org.springframework.messaging.converter.MessageConversionException.class,
+                org.springframework.dao.DataIntegrityViolationException.class,
+                org.hibernate.exception.ConstraintViolationException.class
+        );
+
+        return handler;
+    }
+}

src/main/java/com/isums/maintainservice/domains/events/JobCreatedEvent.java

@@ -1,5 +1,10 @@
 package com.isums.maintainservice.domains.events;
 
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -11,10 +16,40 @@
 @Builder
 @NoArgsConstructor
 @AllArgsConstructor
+@JsonIgnoreProperties(ignoreUnknown = true)
+@JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.ANY,
+        getterVisibility = JsonAutoDetect.Visibility.NONE,
+        isGetterVisibility = JsonAutoDetect.Visibility.NONE,
+        setterVisibility = JsonAutoDetect.Visibility.NONE)
 public class JobCreatedEvent {
+    @JsonProperty("referenceId")
     private UUID referenceId;
+
+    @JsonProperty("houseId")
     private UUID houseId;
+
+    @JsonProperty("referenceType")
     private String referenceType;
+
+    @JsonProperty("type")
     private String type;
+
+    @JsonProperty("messageId")
     private String messageId;
+
+    @JsonCreator
+    public static JobCreatedEvent fromJson(
+            @JsonProperty("referenceId") UUID referenceId,
+            @JsonProperty("houseId") UUID houseId,
+            @JsonProperty("referenceType") String referenceType,
+            @JsonProperty("type") String type,
+            @JsonProperty("messageId") String messageId) {
+        return JobCreatedEvent.builder()
+                .referenceId(referenceId)
+                .houseId(houseId)
+                .referenceType(referenceType)
+                .type(type)
+                .messageId(messageId)
+                .build();
+    }
 }

src/main/java/com/isums/maintainservice/infrastructures/listeners/JobScheduledEventListeners.java

@@ -7,6 +7,8 @@
 import com.isums.maintainservice.domains.events.JobEvent;
 import com.isums.maintainservice.infrastructures.abstracts.InspectionJobService;
 import com.isums.maintainservice.infrastructures.abstracts.MaintenanceJobService;
+import common.kafkas.IdempotencyService;
+import common.kafkas.KafkaListenerHelper;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -23,10 +25,14 @@
 @Slf4j
 public class JobScheduledEventListeners {
 
+    private static final String JOB_CREATED_DLQ_TOPIC = "job.created-dlq";
+
     private final MaintenanceJobService maintenanceJobService;
     private final InspectionJobService inspectionJobService;
     private final ObjectMapper objectMapper;
     private final KafkaTemplate<String, Object> kafka;
+    private final IdempotencyService idempotencyService;
+    private final KafkaListenerHelper kafkaHelper;
 
     @KafkaListener(topics = "job.scheduled", groupId = "maintenance-group")
     public void handleScheduled(ConsumerRecord<String, String> record, Acknowledgment ack) {
@@ -170,11 +176,35 @@ public void handleAssigned(ConsumerRecord<String, String> record, Acknowledgment
 
     @KafkaListener(topics = "job.created", groupId = "maintenance-group")
     public void handleJobCreated(ConsumerRecord<String, String> record, Acknowledgment ack) {
+        String messageId = kafkaHelper.extractMessageId(record);
+        kafkaHelper.setupMDC(record, messageId);
         try {
-            JobCreatedEvent event = objectMapper.readValue(
-                    record.value(), JobCreatedEvent.class);
+            if (idempotencyService.isDuplicate(messageId)) {
+                log.info("[Maintenance] job.created duplicate skipped messageId={}", messageId);
+                ack.acknowledge();
+                return;
+            }
+
+            JobCreatedEvent event = parseJobCreatedEvent(record);
+            if (event == null) {
+                publishToDlq(record, "PARSE_FAILED");
+                ack.acknowledge();
+                return;
+            }
 
             if (!"INSPECTION".equals(event.getReferenceType())) {
+                log.debug("[Maintenance] job.created not INSPECTION (={}), skip messageId={}",
+                        event.getReferenceType(), messageId);
+                idempotencyService.markProcessed(messageId);
+                ack.acknowledge();
+                return;
+            }
+
+            if (event.getType() == null || event.getType().isBlank()
+                    || event.getHouseId() == null || event.getReferenceId() == null) {
+                log.error("[Maintenance] job.created missing required fields type={} houseId={} refId={} messageId={}",
+                        event.getType(), event.getHouseId(), event.getReferenceId(), messageId);
+                publishToDlq(record, "MISSING_REQUIRED_FIELDS");
                 ack.acknowledge();
                 return;
             }
@@ -191,16 +221,70 @@ public void handleJobCreated(ConsumerRecord<String, String> record, Acknowledgme
                             .messageId(UUID.randomUUID().toString())
                             .build());
 
+            idempotencyService.markProcessed(messageId);
             ack.acknowledge();
-            log.info("[Maintenance] InspectionJob created id={} type={} contractId={}",
-                    job.id(), event.getType(), event.getReferenceId());
+            log.info("[Maintenance] InspectionJob created inspectionId={} type={} contractId={} messageId={}",
+                    job.id(), event.getType(), event.getReferenceId(), messageId);
 
-        } catch (com.fasterxml.jackson.core.JsonProcessingException e) {
-            log.error("[Maintenance] Deserialize failed raw={}: {}", record.value(), e.getMessage());
-            ack.acknowledge();
         } catch (Exception e) {
-            log.error("[Maintenance] handleJobCreated failed: {}", e.getMessage(), e);
+            log.error("[Maintenance] handleJobCreated failed messageId={} — will retry: {}",
+                    messageId, e.getMessage(), e);
             throw new RuntimeException(e);
+        } finally {
+            kafkaHelper.clearMDC();
+        }
+    }
+
+    private JobCreatedEvent parseJobCreatedEvent(ConsumerRecord<String, String> record) {
+        try {
+            JobCreatedEvent event = objectMapper.readValue(record.value(), JobCreatedEvent.class);
+            if (event != null && event.getType() != null) return event;
+        } catch (com.fasterxml.jackson.core.JsonProcessingException e) {
+            log.warn("[Maintenance] Strict JSON parse failed, fallback to Map: {}", e.getMessage());
+        }
+        try {
+            java.util.Map<String, Object> raw = objectMapper.readValue(
+                    record.value(),
+                    new com.fasterxml.jackson.core.type.TypeReference<java.util.Map<String, Object>>() {});
+            return JobCreatedEvent.builder()
+                    .referenceId(parseUuid(raw.get("referenceId")))
+                    .houseId(parseUuid(raw.get("houseId")))
+                    .referenceType(asString(raw.get("referenceType")))
+                    .type(asString(raw.get("type")))
+                    .messageId(asString(raw.get("messageId")))
+                    .build();
+        } catch (Exception e) {
+            log.error("[Maintenance] Map fallback parse failed raw={}: {}", record.value(), e.getMessage());
+            return null;
+        }
+    }
+
+    private static UUID parseUuid(Object value) {
+        if (value == null) return null;
+        try {
+            return UUID.fromString(value.toString());
+        } catch (IllegalArgumentException e) {
+            return null;
+        }
+    }
+
+    private static String asString(Object value) {
+        return value == null ? null : value.toString();
+    }
+
+    private void publishToDlq(ConsumerRecord<String, String> record, String reason) {
+        try {
+            kafka.send(JOB_CREATED_DLQ_TOPIC, record.key(), record.value())
+                    .whenComplete((r, ex) -> {
+                        if (ex != null) {
+                            log.error("[Maintenance] DLQ publish failed reason={}: {}", reason, ex.toString());
+                        } else {
+                            log.warn("[Maintenance] Routed to DLQ reason={} originalOffset={} key={}",
+                                    reason, record.offset(), record.key());
+                        }
+                    });
+        } catch (Exception e) {
+            log.error("[Maintenance] DLQ publish threw reason={}: {}", reason, e.getMessage(), e);
         }
     }
 }

src/main/java/com/isums/maintainservice/services/InspectionJobServiceImpl.java

@@ -93,8 +93,8 @@ public InspectionDto create(String creatorKeycloakId, CreateInspectionRequest re
     public InspectionDto createFromEvent(JobCreatedEvent event) {
         try {
             String noteVi = event.getType().equals("CHECK_IN")
-                    ? "Pre-handover house inspection"
-                    : "End-of-contract house inspection";
+                    ? "Kiểm tra bàn giao nhà trước khi khách vào ở"
+                    : "Kiểm tra trả nhà khi kết thúc hợp đồng";
             TranslationMap noteTranslations = translationAutoFillService.complete(noteVi, DEFAULT_LANGUAGE);
 
             InspectionJob job = InspectionJob.builder()