From 3b0dba7ce5a55d5e3f1eddd346a8bb1c9e3449df Mon Sep 17 00:00:00 2001 From: hoangtuzami Date: Sat, 16 May 2026 11:17:50 +0700 Subject: [PATCH] fix(consumer): harden job event listeners with idempotency, DLQ, and explicit Jackson mapping - Add idempotency check and MDC setup in handleJobCreated; route poison messages to job.created-dlq with reason codes (PARSE_FAILED / MISSING_REQUIRED_FIELDS). - Add 2-tier parsing fallback in parseJobCreatedEvent so cross-service Jackson version drift no longer drops events. - Add explicit @JsonProperty / @JsonIgnoreProperties / @JsonAutoDetect / @JsonCreator to JobCreatedEvent to stop silent null fields from polymorphic typing. - Localize default inspection note to Vietnamese. - Add KafkaConsumerConfig with DLT recoverer + producer factories. Co-Authored-By: Claude Opus 4.7 (1M context) --- .../configurations/KafkaConsumerConfig.java | 70 ++++++++++++ .../domains/events/JobCreatedEvent.java | 35 ++++++ .../listeners/JobScheduledEventListeners.java | 100 ++++++++++++++++-- .../services/InspectionJobServiceImpl.java | 4 +- 4 files changed, 199 insertions(+), 10 deletions(-) create mode 100644 src/main/java/com/isums/maintainservice/configurations/KafkaConsumerConfig.java diff --git a/src/main/java/com/isums/maintainservice/configurations/KafkaConsumerConfig.java b/src/main/java/com/isums/maintainservice/configurations/KafkaConsumerConfig.java new file mode 100644 index 0000000..8a7cc33 --- /dev/null +++ b/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 objectKafkaTemplate() { + Map 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 dltKafkaTemplate() { + Map 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 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; + } +} diff --git a/src/main/java/com/isums/maintainservice/domains/events/JobCreatedEvent.java b/src/main/java/com/isums/maintainservice/domains/events/JobCreatedEvent.java index c48e185..21a2986 100644 --- a/src/main/java/com/isums/maintainservice/domains/events/JobCreatedEvent.java +++ b/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(); + } } diff --git a/src/main/java/com/isums/maintainservice/infrastructures/listeners/JobScheduledEventListeners.java b/src/main/java/com/isums/maintainservice/infrastructures/listeners/JobScheduledEventListeners.java index c6e1d86..385ba09 100644 --- a/src/main/java/com/isums/maintainservice/infrastructures/listeners/JobScheduledEventListeners.java +++ b/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 kafka; + private final IdempotencyService idempotencyService; + private final KafkaListenerHelper kafkaHelper; @KafkaListener(topics = "job.scheduled", groupId = "maintenance-group") public void handleScheduled(ConsumerRecord record, Acknowledgment ack) { @@ -170,11 +176,35 @@ public void handleAssigned(ConsumerRecord record, Acknowledgment @KafkaListener(topics = "job.created", groupId = "maintenance-group") public void handleJobCreated(ConsumerRecord 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 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 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 raw = objectMapper.readValue( + record.value(), + new com.fasterxml.jackson.core.type.TypeReference>() {}); + 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 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); } } } \ No newline at end of file diff --git a/src/main/java/com/isums/maintainservice/services/InspectionJobServiceImpl.java b/src/main/java/com/isums/maintainservice/services/InspectionJobServiceImpl.java index a53d2a8..a63897d 100644 --- a/src/main/java/com/isums/maintainservice/services/InspectionJobServiceImpl.java +++ b/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()