From 005e228bba833bdc23e491dc0ce4039d3d128d91 Mon Sep 17 00:00:00 2001 From: Mikhail Trofimov Date: Fri, 10 Jul 2026 09:52:39 +0300 Subject: [PATCH] Get rid of InquiryDateDeserializer; Changed ack mode --- .../config/kafka/KafkaConfig.java | 14 +++++++-- .../kafka/InquiryDateDeserializer.java | 30 ------------------- .../kafka/PackageMessage.java | 10 +++---- .../monitoring/KafkaMonitoringService.java | 1 + 4 files changed, 16 insertions(+), 39 deletions(-) delete mode 100644 src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/InquiryDateDeserializer.java diff --git a/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java b/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java index 8faf0ef..5d20195 100644 --- a/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java +++ b/src/main/java/ru/nbch/credit_tracker/config/kafka/KafkaConfig.java @@ -1,5 +1,7 @@ package ru.nbch.credit_tracker.config.kafka; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; import org.apache.kafka.clients.consumer.ConsumerConfig; import org.apache.kafka.common.serialization.StringDeserializer; import org.springframework.context.annotation.Bean; @@ -35,12 +37,13 @@ public class KafkaConfig { new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory); factory.setBatchListener(true); - factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL); + factory.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE); return factory; } @Bean + //todo надо проверить конфигурацию внимательно public ConsumerFactory consumerFactory(CreditTrackerSettings settings) { KafkaConsumerSettings kafkaSettings = settings.getKafkaConsumerSettings(); Map props = new HashMap<>(); @@ -50,7 +53,12 @@ public class KafkaConfig { props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, kafkaSettings.maxPollRecords()); props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, kafkaSettings.maxPollIntervalMs()); - JsonDeserializer jsonDeserializer = new JsonDeserializer<>(PackageMessage.class); + ObjectMapper objectMapper = new ObjectMapper(); + objectMapper.registerModule(new JavaTimeModule()); + + JsonDeserializer jsonDeserializer = new JsonDeserializer<>( + PackageMessage.class, + objectMapper); jsonDeserializer.addTrustedPackages("*"); jsonDeserializer.setUseTypeHeaders(false); @@ -61,7 +69,7 @@ public class KafkaConfig { props, new StringDeserializer(), jsonDeserializer - ); + ); } } diff --git a/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/InquiryDateDeserializer.java b/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/InquiryDateDeserializer.java deleted file mode 100644 index 2c66462..0000000 --- a/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/InquiryDateDeserializer.java +++ /dev/null @@ -1,30 +0,0 @@ -package ru.nbch.credit_tracker.entities.external_request.kafka; - -import com.fasterxml.jackson.core.JsonParser; -import com.fasterxml.jackson.core.JsonToken; -import com.fasterxml.jackson.databind.DeserializationContext; -import com.fasterxml.jackson.databind.JsonDeserializer; - -import java.io.IOException; -import java.time.LocalDateTime; -import java.util.List; - -public class InquiryDateDeserializer extends JsonDeserializer { - @Override - public LocalDateTime deserialize(JsonParser p, DeserializationContext ctxt) throws IOException { - if (p.currentToken() == JsonToken.START_ARRAY) { - List dateParts = p.readValueAs(List.class); - if (dateParts.size() >= 6) { - int year = dateParts.get(0); - int month = dateParts.get(1); - int day = dateParts.get(2); - int hour = dateParts.get(3); - int minute = dateParts.get(4); - int second = dateParts.get(5); - int nano = dateParts.size() > 6 ? dateParts.get(6) : 0; - return LocalDateTime.of(year, month, day, hour, minute, second, nano); - } - } - return null; - } -} diff --git a/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/PackageMessage.java b/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/PackageMessage.java index 6b04322..83c3d8b 100644 --- a/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/PackageMessage.java +++ b/src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/PackageMessage.java @@ -1,6 +1,7 @@ package ru.nbch.credit_tracker.entities.external_request.kafka; -import com.fasterxml.jackson.databind.annotation.JsonDeserialize; +import com.fasterxml.jackson.annotation.JsonFormat; +import com.fasterxml.jackson.annotation.JsonIgnoreProperties; import lombok.AllArgsConstructor; import lombok.Getter; import lombok.NoArgsConstructor; @@ -13,10 +14,11 @@ import java.util.List; @NoArgsConstructor @Getter @Setter +@JsonIgnoreProperties(ignoreUnknown = true) public class PackageMessage { private Integer fid; private String memberCode; - @JsonDeserialize(using = InquiryDateDeserializer.class) + @JsonFormat(shape = JsonFormat.Shape.ARRAY) private LocalDateTime inquiryDate; private String inquiryPurpose; private Long inquiryAmount; @@ -24,10 +26,6 @@ public class PackageMessage { private Integer inquiryType; private List persons; - public boolean isNaturalPerson() { - return persons != null && !persons.isEmpty(); - } - @AllArgsConstructor @NoArgsConstructor @Getter diff --git a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java index c041be3..9a4a1a7 100644 --- a/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java +++ b/src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java @@ -105,6 +105,7 @@ public class KafkaMonitoringService { private void process(List messages) { Map> partition = messages.stream() + .filter(Objects::nonNull) .filter(this::filterMessages) .collect(Collectors.partitioningBy(m -> m.getPersons() != null && !m.getPersons().isEmpty()));