Merge pull request #3767 from ingef/fix/race-condition-cce-load-storage

Fix: Race condition on loadStorage

Author: thoniTUB

backend/src/main/java/com/bakdata/conquery/io/storage/Store.java

@@ -14,6 +14,11 @@ public interface Store<KEY, VALUE> extends ManagedStore {
 
 	VALUE get(KEY key);
 
+	/**
+	 * Iterate over all key-value pairs in the store.
+	 * @param consumer Key-value consumer.
+	 * @implSpec    The consumer must be thread-safe
+	 */
 	IterationStatistic forEach(StoreEntryConsumer<KEY, VALUE> consumer);
 
 	// TODO: 08.01.2020 fk: Is this still necessary? The implementation in XodusStore uses different methods that in our context don't act differently.

backend/src/main/java/com/bakdata/conquery/io/storage/xodus/stores/CachedStore.java

@@ -5,6 +5,7 @@
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.LongAdder;
 import java.util.stream.Stream;
 
@@ -153,7 +154,7 @@ public void loadData() {
 
 		final Stopwatch timer = Stopwatch.createStarted();
 
-		final Set<KEY> dupes = new HashSet<>();
+		final Set<KEY> dupes = Collections.newSetFromMap(new ConcurrentHashMap<>());
 
 		store.forEach((key, value, size) -> {
 			try {