Get rid of InquiryDateDeserializer; Changed ack mode

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<String, PackageMessage> consumerFactory(CreditTrackerSettings settings) {
         KafkaConsumerSettings kafkaSettings = settings.getKafkaConsumerSettings();
         Map<String, Object> 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<PackageMessage> jsonDeserializer = new JsonDeserializer<>(PackageMessage.class);
+        ObjectMapper objectMapper = new ObjectMapper();
+        objectMapper.registerModule(new JavaTimeModule());
+
+        JsonDeserializer<PackageMessage> jsonDeserializer = new JsonDeserializer<>(
+                PackageMessage.class,
+                objectMapper);
         jsonDeserializer.addTrustedPackages("*");
         jsonDeserializer.setUseTypeHeaders(false);
 
@@ -61,7 +69,7 @@ public class KafkaConfig {
                 props,
                 new StringDeserializer(),
                 jsonDeserializer
-                );
+        );
     }
 
 }

src/main/java/ru/nbch/credit_tracker/entities/external_request/kafka/InquiryDateDeserializer.java

@@ -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<LocalDateTime> {
-    @Override
-    public LocalDateTime deserialize(JsonParser p, DeserializationContext ctxt) throws IOException {
-        if (p.currentToken() == JsonToken.START_ARRAY) {
-            List<Integer> 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;
-    }
-}

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<Person> persons;
 
-    public boolean isNaturalPerson() {
-        return persons != null && !persons.isEmpty();
-    }
-
     @AllArgsConstructor
     @NoArgsConstructor
     @Getter

src/main/java/ru/nbch/credit_tracker/service/mode/monitoring/KafkaMonitoringService.java

@@ -105,6 +105,7 @@ public class KafkaMonitoringService {
 
     private void process(List<PackageMessage> messages) {
         Map<Boolean, List<PackageMessage>> partition = messages.stream()
+                .filter(Objects::nonNull)
                 .filter(this::filterMessages)
                 .collect(Collectors.partitioningBy(m -> m.getPersons() != null && !m.getPersons().isEmpty()));