Merge pull request #84

FEAT : CDE-403-sur-un-gros-fichier-le-bad-ecrase-son-contenu-au-fil-d…

src/main/java/fr/abes/bestppn/kafka/TopicConsumer.java

@@ -107,6 +107,8 @@ public void kbartFromkafkaListener(ConsumerRecord<String, String> ligneKbart) {
                     log.error(e.getMessage());
                     workInProgress.get(filename).addLineKbartToMailAttachementWithErrorMessage(ligneKbartDto, e.getMessage());
                     workInProgress.get(filename).addNbLinesWithErrorsInExecutionReport();
+                } catch (InterruptedException e) {
+                    throw new RuntimeException(e);
                 } finally {
                     //on ne décrémente pas le nb de thread si l'objet de suivi a été supprimé après la production des messages dans le second topic
                     if (workInProgress.get(filename) != null)
@@ -123,7 +125,7 @@ public void kbartFromkafkaListener(ConsumerRecord<String, String> ligneKbart) {
     }
 
 
-    private void handleFichier(String filename) {
+    private void handleFichier(String filename) throws InterruptedException {
         //on attend que l'ensemble des threads aient terminé de travailler avant de lancer le commit
         do {
             try {
@@ -151,13 +153,14 @@ private void handleFichier(String filename) {
             log.error("Le nom du fichier " + filename + " n'est pas correct. " + e);
             emailService.sendProductionErrorEmail(filename, e.getMessage());
         } finally {
+            Thread.sleep(5000); // pour attendre que tous les threads de best-ppn-api aient terminé leurs traitements
             log.info("Traitement terminé pour fichier " + filename + " / nb lignes " + workInProgress.get(filename).incrementNbLignesTraiteesAndGet());
             workInProgress.remove(filename);
         }
     }
 
     @KafkaListener(topics = {"${topic.name.source.kbart.errors}"}, groupId = "${topic.groupid.source.errors}", containerFactory = "kafkaKbartListenerContainerFactory")
-    public void errorsListener(ConsumerRecord<String, String> error) {
+    public void errorsListener(ConsumerRecord<String, String> error) throws InterruptedException {
         log.error(error.value());
         String filename = error.key();
         do {