Synchronise recreate in-memory method
@@ -439,7 +439,7 @@ public class AggregationRuntime implements MemoryCalculable {
incrementalDataPurging.executeIncrementalDataPurging();
}
- public void recreateInMemoryData(boolean isFirstEventArrived) {
+ public synchronized void recreateInMemoryData(boolean isFirstEventArrived) {
// State only updated when first event arrives to IncrementalAggregationProcessor
if (isFirstEventArrived) {
this.isFirstEventArrived = true;
@@ -448,10 +448,8 @@ public class AggregationRuntime implements MemoryCalculable {
durationIncrementalExecutorEntry.getValue().setProcessingExecutor(true);
}
}
- synchronized (this) {
this.recreateInMemoryData.recreateInMemoryData();
}
- }
public void processEvents(ComplexEventChunk<StreamEvent> streamEventComplexEventChunk) {
incrementalExecutorMap.get(incrementalDurations.get(0)).execute(streamEventComplexEventChunk);
