title: Concurrent access to lists
body: <p>I needed in my project something that would group calls to my server because there are some "batch" routes that accept a list of id instead of a simple id.</p>
  
  <p>My idea was to build something with a pool that would be emptied once in a while.</p>
  
  <p>So I create this <code>AbstractRepository</code> (not sure if it's well named) :</p>
  
  <pre><code>public abstract class AbstractRepository&lt;T&gt;{
  
      protected Context c;
  
      //interval between 2 queue emptying
      private final int POOL_DELAY = 200;
      protected int downloadingTaskCount = 0;
  
      final protected ArrayMap&lt;String, T&gt; memCache = new ArrayMap&lt;&gt;();
      final protected HashSet&lt;String&gt; queue = new HashSet&lt;&gt;();
  
      final protected ArrayMap&lt;String, List&lt;FetchedInterface&lt;T&gt;&gt;&gt; callbackMap = new ArrayMap&lt;&gt;();
  
      final protected List&lt;PoolEmptiedInterface&gt; emptinessWatchers = new ArrayList&lt;&gt;();
  
  
      protected AbstractRepository(Context c) {
          handler.postDelayed(downloadRoutine, POOL_DELAY);
          this.c = c;
      }
  
      public void cache(String id) {
          if (!memCache.containsKey(memCache)) {
              synchronized (queue) {
                  queue.add(id);
              }
          }
      }
  
      public void getCache(String id, FetchedInterface&lt;T&gt; callback) {
  
          if (memCache.containsKey(id)) {
              callback.fetched(memCache.get(id));
          } else {
  
              synchronized (callbackMap) {
                  if (!callbackMap.containsKey(id)) {
                      callbackMap.put(id, new ArrayList&lt;FetchedInterface&lt;T&gt;&gt;());
                  }
                  callbackMap.get(id).add(callback);
              }
  
              synchronized (queue) {
                  queue.add(id);
              }
  
          }
      }
  
      public void getCacheIdObj(List&lt;IdObject&gt; idsObj, final ListFetchedInterface&lt;T&gt; callback) {
          ArrayList&lt;String&gt; ids = new ArrayList&lt;&gt;();
          for (IdObject idObj : idsObj) {
              ids.add(idObj.getId());
          }
          getCache(ids, callback);
      }
  
      public void getCache(List&lt;String&gt; ids, final ListFetchedInterface&lt;T&gt; callback) {
          final CountDownLatch countDownLatch = new CountDownLatch(ids.size());
          final ArrayList&lt;T&gt; array = new ArrayList&lt;&gt;();
          for (String id : ids) {
              getCache(id, new FetchedInterface&lt;T&gt;() {
                  @Override
                  public void fetched(T item) {
                      array.add(item);
                      countDownLatch.countDown();
                  }
              });
  
          }
  
          new Thread(new Runnable() {
              @Override
              public void run() {
                  try {
                      countDownLatch.await();
                      callback.fetched(array);
                  } catch (InterruptedException e) {
                      e.printStackTrace();
                  }
              }
          }).start();
  
      }
  
  
      /**
       * Exists for threads that want to be notified that the user queue has been flushed.
       */
      public void getNotifiedWhenQueueIsEmptied(PoolEmptiedInterface&lt;T&gt; callback) {
          if (downloadingTaskCount == 0 &amp;&amp; queue.isEmpty()) {
              callback.poolEmpty();
          } else {
              synchronized (emptinessWatchers) {
                  emptinessWatchers.add(callback);
              }
          }
      }
  
      protected void doIt(
              final HashSet&lt;String&gt; processingQueue) {
      }
  
      /**
       * Pool Loop
       */
      Handler handler = new Handler();
  
      private Runnable downloadRoutine = new Runnable() {
          @Override
          public void run() {
  
              if (!queue.isEmpty()) {
                  final HashSet&lt;String&gt; processingQueue = new HashSet&lt;&gt;();
  
                  synchronized (queue) {
                      processingQueue.addAll(queue);
                      queue.clear();
                  }
                  downloadingTaskCount++;
                  doIt(processingQueue);
              }
  
              handler.postDelayed(downloadRoutine, POOL_DELAY);
          }
      };
  </code></pre>
  
  <p>And one of its children <code>UserRepository</code> </p>
  
  <pre><code>public class UserRepository extends AbstractRepository&lt;UserCache&gt; {
  
      private static volatile UserRepository instance;
  
      public static UserRepository getInstance(Context c) {
          synchronized (UserRepository.class) {
              if (instance == null) {
                  instance = new UserRepository(c);
              }
              return instance;
          }
      }
  
      private UserRepository(Context c) {
          super(c);
      }
  
      @Override
      protected void doIt(final HashSet&lt;String&gt; processingQueue) {
          Api.getInstance().backend.getUsersCache(new IdListArguments(new ArrayList&lt;&gt;(processingQueue)))
                  .enqueue(new Callback&lt;Map&lt;String, UserCache&gt;&gt;() {
                      @Override
                      public void onResponse(Call&lt;Map&lt;String, UserCache&gt;&gt; call, Response&lt;Map&lt;String, UserCache&gt;&gt; responseParent) {
                          Map&lt;String, UserCache&gt; response = responseParent.body();
                          Iterator&lt;String&gt; it = processingQueue.iterator();
                          while (it.hasNext()) {
  
                              String id = it.next();
                              if (response.containsKey(id)) {
                                  memCache.put(id, response.get(id));
                                  if (callbackMap.containsKey(id)) {
                                      for (FetchedInterface callback : callbackMap.get(id)) {
                                          callback.fetched(response.get(id));
                                      }
                                  }
                                  it.remove();
                              }
                          }
  
                          for (PoolEmptiedInterface watcher : emptinessWatchers) {
                              watcher.poolEmpty();
                          }
                          downloadingTaskCount--;
                          queue.addAll(processingQueue);
                      }
  
                      @Override
                      public void onFailure(Call&lt;Map&lt;String, UserCache&gt;&gt; call, Throwable t) {
                          queue.addAll(processingQueue);
                      }
                  });
      }
  }
  </code></pre>
  
  <p>My Exception :</p>
  
  <pre><code>Java.util.ConcurrentModificationException
    at java.util.ArrayList$ArrayListIterator.next(ArrayList.java:573)
    at com.m360.android.domain_layer.interactors.MemberInteractor.constructPagerMemberUsers(MemberInteractor.java:116)
    at com.m360.android.domain_layer.interactors.MemberInteractor.access$000(MemberInteractor.java:29)
    at com.m360.android.domain_layer.interactors.MemberInteractor$2$1.fetched(MemberInteractor.java:64)
    at com.m360.android.datalayer.repositories.AbstractRepository$2.run(AbstractRepository.java:99)
  </code></pre>
  
  <p>And MemberInteractor only contains static methods, the crash appears using the following 2 :</p>
  
  <pre><code>public static void getGroupUsers(String id, final Context c, final ErrorDisplayerInterface i, final MemberInteractorCallback callback) {
      GroupRepository.getInstance(c).getCache(id, new AbstractRepository.FetchedInterface&lt;Group&gt;() {
          @Override
          public void fetched(Group item) {
              UserRepository userRepositoryNew = UserRepository.getInstance(c);
              userRepositoryNew.getCache(new ArrayList&lt;&gt;(item.getUsers()), new AbstractRepository.ListFetchedInterface&lt;UserCache&gt;() {
                  @Override
                  public void fetched(List&lt;UserCache&gt; items) {
                      callback.onFinish(constructPagerMemberUsers(items));
                  }
              });
          }
      });
  }
  
  
  private static List&lt;PagerMemberUser&gt; constructPagerMemberUsers(final List&lt;UserCache&gt; items) {
  
      final List&lt;PagerMemberUser&gt; users = new ArrayList&lt;&gt;();
      for (UserCache item : items) {
          users.add(new PagerMemberUser(item));
      }
      return users;
  }
  </code></pre>
  
  <p>I'm sorry there is a lot of code but I think all of it is relevant to my question.</p>
  
  <p>So what's happening there ? I don't see the problem.</p>