bugfix: when building materialized-view, if taskCount>1, may cause concurrentModificationException

[pzhdfy]

In checkSegmentsAndSubmitTasks:

for (Map.Entry<Interval, HadoopIndexTask> entry : runningTasks.entrySet()) {
        Optional<TaskStatus> taskStatus = taskStorage.getStatus(entry.getValue().getId());
        if (!taskStatus.isPresent() || !taskStatus.get().isRunnable()) {
          runningTasks.remove(entry.getKey());
          runningVersion.remove(entry.getKey());
        }
      }

if taskCount>1, while iterating runningTasks and removing entry from runningTasks will happen concurrent, this will cause concurrentModificationException.

use ConcurrentHashMap instead of HashMap for runningTasks, will fix

[jihoonson]

@pzhdfy thanks for the patch and sorry for late review. Do you know what threads can read and remove at the same time? It's currently guarded by taskLock.

    synchronized (taskLock) {
      for (Map.Entry<Interval, HadoopIndexTask> entry : runningTasks.entrySet()) {
        Optional<TaskStatus> taskStatus = taskStorage.getStatus(entry.getValue().getId());
        if (!taskStatus.isPresent() || !taskStatus.get().isRunnable()) {
          runningTasks.remove(entry.getKey());
          runningVersion.remove(entry.getKey());
        }
      }
      if (runningTasks.size() == maxTaskCount) {
        //if the number of running tasks reach the max task count, supervisor won't submit new tasks.
        return;
      }
      Pair<SortedMap<Interval, String>, Map<Interval, List<DataSegment>>> toBuildIntervalAndBaseSegments = checkSegments();
      SortedMap<Interval, String> sortedToBuildVersion = toBuildIntervalAndBaseSegments.lhs;
      Map<Interval, List<DataSegment>> baseSegments = toBuildIntervalAndBaseSegments.rhs;
      missInterval = sortedToBuildVersion.keySet();
      submitTasks(sortedToBuildVersion, baseSegments);
    }

[pzhdfy]

@jihoonson

Yes, this is guarded by taskLock .

This is just the WRONG USE of hashmap.

if we remove an entry from a hashmap by remove(key), while iterating it, this will throw a concurrentModificationException, even in a single thread.

ref:https://stackoverflow.com/questions/602636/concurrentmodificationexception-and-a-hashmap

for (Map.Entry<Interval, HadoopIndexTask> entry : runningTasks.entrySet()) {
        Optional<TaskStatus> taskStatus = taskStorage.getStatus(entry.getValue().getId());
        if (!taskStatus.isPresent() || !taskStatus.get().isRunnable()) {
          runningTasks.remove(entry.getKey());
          runningVersion.remove(entry.getKey());
        }
      }

[jihoonson]

Oh I see. You’re right. Thanks.

I think it would be better to first find the entries to be removed, and then remove then all at once after iteration. I think this is better than using ConcurrentMap because there’s no concurrency issue. What do you think?

Also, all bug fix PRs should have unit tests. Would you please add one?

[gianm]

Or how about using .remove() on the iterator? I think if these maps are not actually going to be used from multiple threads, then we don't need to make them ConcurrentMaps.

[jihoonson]

Hi @pzhdfy, were you able to check this issue again by any chance?

[pzhdfy]

@jihoonson
I have updated the pr,
removing entry after iteration instead of using ConcurrentMap, and add unit test

[jihoonson]

@pzhdfy thank you for the update! Please check my comment.

[jihoonson]

LGTM. @pzhdfy thank you for the quick fix!