Привет!
Можно вопрос к знатокам многопоточности?
Есть такой код:
Map<String, List<Data>> cache = new HashMap<>();
String currentBatch;      
public void onNext(Data data) {         
    String batchId = data.getBatchId();          
    synchronized (this) {             
         if (cache.get(batchId) == null) {                 
             CompletableFuture.runAsync(() -> processBatch(currentBatch, cache.get(currentBatch));
             currentBatch = batchId;       
             cache.put(batchId, new ArrayList<>());
         }
     }         
    cache.get(currentBatch).add(data); 
}
Почему-то так происходит, что периодически в функцию processBatch передается уже изменённое значение currentButch. Пытался как-то зафиксировать порядок вызова функции и обновления значения currentBatch, вставляя между ними, например, инкремент AtomicInteger - не помогает.