Является ли следующий код безопасным для потоков

У меня есть сценарий, в котором мне нужно поддерживать карту, которая может быть заполнена несколькими потоками, каждый из которых изменяет соответствующий список (уникальный идентификатор / ключ является именем потока), и когда размер списка для потока превышает фиксированный размер пакета, мы должны сохранять записи в БД.

Пример кода ниже:

private volatile ConcurrentHashMap<String, List<T>>  instrumentMap = new ConcurrentHashMap<String, List<T>>();
private ReadWriteLock lock ;

public void addAll(List<T> entityList, String threadName) {
    try {
        lock.readLock().lock();
        List<T> instrumentList = instrumentMap.get(threadName);
        if (instrumentList == null) {
            instrumentList = new ArrayList<T>(batchSize);
            instrumentMap.put(threadName, instrumentList);
        }

        if (instrumentList.size() >= batchSize -1){
            instrumentList.addAll(entityList);
            recordSaver.persist(instrumentList); 
            instrumentList.clear();
        } else {
            instrumentList.addAll(entityList);  
        }
    } finally {
        lock.readLock().unlock();
    }

}

Через каждые 2 минуты запускается еще один отдельный поток для сохранения всех записей в Map (чтобы убедиться, что у нас что-то сохраняется через каждые 2 минуты и размер карты не становится слишком большим), и когда он запускается, он блокирует все другие потоки (проверьте readLock и writeLock нас, где writeLock имеет более высокий приоритет)

if (//Some condition) {
                    Thread.sleep(//2 minutes);
                    aggregator.getLock().writeLock().lock();
                    List<T> instrumentList = instrumentMap .values().stream().flatMap(x->x.stream()).collect(Collectors.toList());
                    if (instrumentList.size() > 0) {

                        saver.persist(instrumentList);
                        instrumentMap .values().parallelStream().forEach(x -> x.clear());
                    aggregator.getLock().writeLock().unlock();
                }

Это решение работает нормально почти для каждого сценария, который мы тестировали, за исключением того, что иногда мы видим, что некоторые записи пропали, то есть вообще не сохранялись, хотя они были добавлены в Map.

У меня вопрос, в чем проблема с этим кодом? ConcurrentHashMap - не лучшее решение здесь? Есть ли здесь проблемы с использованием блокировки чтения / записи? Следует ли мне использовать последовательную обработку?