Je vois que vous utilisez le parallelStream de ConcurrentHashMap dans un verrou. Je ne connais pas la prise en charge des flux Java 8+, mais une recherche rapide montre que

1. ConcurrentHashMap est une structure de données complexe, qui avait autrefois des bogues d'accès concurrentiel
2. Les flux parallèles doivent respecter les restrictions d'utilisation complexes et mal documentées
3. Vous modifiez vos données dans un flux parallèle

Sur la base de ces informations (et de mon détecteur de bogues de concurrence basé sur l'intestin ™), je parie que la suppression de l'appel à parallelStream pourrait améliorer la robustesse de votre code. De plus, comme mentionné par @Slaw, vous devez utiliser HashMap ordinaire à la place de ConcurrentHashMap si toute l'utilisation de instrumentMap est déjà protégée par un verrou.

Bien sûr, puisque vous ne publiez pas le code de recordSaver , il est possible qu'il comporte également des bogues (et pas nécessairement des bogues liés à la concurrence). En particulier, vous devez vous assurer que le code qui lit les enregistrements à partir du stockage persistant - celui que vous utilisez pour détecter la perte d'enregistrements - est sûr, correct et correctement synchronisé avec le reste de votre système (de préférence en utilisant un , base de données SQL standard).

Il semble que ce soit une tentative d'optimisation là où ce n'était pas nécessaire. Dans ce cas, moins c'est plus et plus simple c'est mieux. Dans le code ci-dessous, seuls deux concepts de concurrence sont utilisés: synchronisé pour garantir qu'une liste partagée est correctement mise à jour et final pour garantir que tous les threads voient la même valeur.

import java.util.ArrayList;
import java.util.List;

public class Aggregator<T> implements Runnable {

    private final List<T> instruments = new ArrayList<>();

    private final RecordSaver recordSaver;
    private final int batchSize;


    public Aggregator(RecordSaver recordSaver, int batchSize) {
        super();
        this.recordSaver = recordSaver;
        this.batchSize = batchSize;
    }

    public synchronized void addAll(List<T> moreInstruments) {

        instruments.addAll(moreInstruments);
        if (instruments.size() >= batchSize) {
            storeInstruments();
        }
    }

    public synchronized void storeInstruments() {

        if (instruments.size() > 0) {
            // in case recordSaver works async
            // recordSaver.persist(new ArrayList<T>(instruments));
            // else just:
            recordSaver.persist(instruments);
            instruments.clear();
        }
    }


    @Override
    public void run() {

        while (true) {
            try { Thread.sleep(1L); } catch (Exception ignored) {
                break;
            }
            storeInstruments();
        }
    }


    class RecordSaver {
        void persist(List<?> l) {}
    }

}

La réponse fournie par @Slaw dans les commentaires a fait l'affaire. Nous laissions l'instance instrumentList s'échapper de manière non synchronisée, c'est-à-dire que l'accès / les opérations se déroulent sur la liste sans aucune synchronisation. Corriger la même chose en passant la copie à d'autres méthodes a fait l'affaire.

La ligne de code suivante est celle où ce problème se produisait

recordSaver.persist (instrumentList); instrumentList.clear ();

Ici, nous permettons à l'instance instrumentList de s'échapper de manière non synchronisée, c'est-à-dire qu'elle est passée à une autre classe (recordSaver.persist) sur laquelle elle devait être actionnée, mais nous effaçons également la liste dans la ligne suivante (dans la classe Aggregator) et tout cela se passe de manière non synchronisée. L'état de la liste ne peut pas être prédit dans l'économiseur d'enregistrement ... une erreur vraiment stupide.

Nous avons résolu le problème en passant une copie clonée de instrumentList à la méthode recordSaver.persist (...). De cette façon, instrumentList.clear () n'a aucun effet sur la liste disponible dans recordSaver pour d'autres opérations.