Fixed concurrent modification exception on function worker stop
@@ -58,6 +58,7 @@ import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
+import java.util.TreeMap;
import java.util.concurrent.ConcurrentHashMap;
import java.util.stream.Collectors;
@@ -424,7 +425,10 @@ public class FunctionRuntimeManager implements AutoCloseable{
final String workerId = this.workerConfig.getWorkerId();
Map<String, Assignment> assignments = workerIdToAssignments.get(workerId);
if (assignments != null) {
- assignments.values().forEach(assignment -> {
+ // Take a copy of the map since the stopFunction will modify the same map
+ // and invalidate the iterator
+ Map<String, Assignment> copiedAssignments = new TreeMap<>(assignments);
+ copiedAssignments.values().forEach(assignment -> {
String fullyQualifiedInstanceId = FunctionCommon.getFullyQualifiedInstanceId(assignment.getInstance());
try {
stopFunction(fullyQualifiedInstanceId, false);
