RxJava ConcurrentModificationException

Question

I'm noob in RxJava but someday... ; )

What I have:

• Receiving json with Event Planner Model
• inside this model the are Tasks
• I want to save this tasks into local db but also want add to them parent_id in case of relationship

Can someone with more experience take a look at this code:

    private fun saveEventPlanners(eventPlanners: ArrayList<EventPlanner>) {
    LogMgr.d(TAG, "saveEventPlanners() events:$eventPlanners")

    compositeDisposable.add(wfmStorageDomain.saveEventPlanners(eventPlanners)
            .subscribeOn(rxSchedulers.computation())
            .observeOn(rxSchedulers.computation())
            .subscribe({ saved ->
                LogMgr.v(TAG, "event planners saved successfully, value: $saved")

                val taskList = ArrayList<Task>()

                eventPlanners.forEach { eventPlanner ->

                    if (eventPlanner.status == EventPlanner.EventPlannerStatus.NEW) {
                        FS.get().wfmComponent.getEventPlannerStatusChanger().updateEventPlannersStatus(eventPlanner.event_id!!, EventPlanner.EventPlannerStatus.RECEIVED)
                    }

                    (eventPlanner.tasks as ArrayList<Task>).forEach {
                        val task = Task()
                        task.id = it.id
                        task.status_id = Task.STATUS.NEW
                        task.name = it.name
                        task.end_scenario_id = it.end_scenario_id
                        task.start_scenario_id = it.start_scenario_id
                        task.isBind = it.isBind
                        task.parent_event_planner_id = eventPlanner.event_planner_id

                        taskList.add(task)
                    }
                    compositeDisposable.add(wfmStorageDomain.saveTasks(taskList)
                            .subscribeOn(rxSchedulers.computation())
                            .observeOn(rxSchedulers.computation())
                            .subscribe({
                                LogMgr.d(TAG, "tasks saved successfully = $it")
                            }, {
                                LogMgr.e(TAG, "Error while saving tasks", it)
                            }))

                }
            }, {
                LogMgr.e(TAG, "Error while saving event planners", it)
            }))

    notifyObservers(eventPlanners)
}

I'm getting error in this function:

java.util.ConcurrentModificationException
    at java.util.ArrayList$ArrayListIterator.next(ArrayList.java:573)
    at com.raizlabs.android.dbflow.sql.saveable.ListModelSaver.saveAll(ListModelSaver.java:32)
    at com.raizlabs.android.dbflow.sql.saveable.ListModelSaver.saveAll(ListModelSaver.java:19)
    at com.raizlabs.android.dbflow.structure.ModelAdapter.saveAll(ModelAdapter.java:196)
    at com.raizlabs.android.dbflow.rx2.structure.RXModelAdapter$3.call(RXModelAdapter.java:61)
    at com.raizlabs.android.dbflow.rx2.structure.RXModelAdapter$3.call(RXModelAdapter.java:58)
    at io.reactivex.internal.operators.completable.CompletableFromCallable.subscribeActual(CompletableFromCallable.java:35)
    at io.reactivex.Completable.subscribe(Completable.java:1919)
    at io.reactivex.internal.operators.completable.CompletableSubscribeOn$SubscribeOnObserver.run(CompletableSubscribeOn.java:64)
    at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:579)
    at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:66)
    at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:57)
    at java.util.concurrent.FutureTask.run(FutureTask.java:234)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:153)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:267)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1080)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:573)
    at java.lang.Thread.run(Thread.java:841)

And this is how looks saving into db:

    override fun saveTasks(tasks: ArrayList<Task>): Single<Boolean> {
    LogMgr.d(TAG, "saveTasks() : $tasks")
    return Single.create({ emitter ->
        RXModelAdapter.from(Task::class.java)
                .saveAll(tasks)
                .subscribeOn(getSubscriptionScheduler())
                .subscribe({
                    LogMgr.d(TAG, "saveTasks() onComplete")
                    emitter.onSuccess(true)

                }, {
                    LogMgr.e(TAG, "saveTasks() onError ", it)
                    emitter.onError(it)
                })
    })
}

  private fun getSubscriptionScheduler(): Scheduler {
    return FS.get().commonComponent.rxSchedulers.get().sqlite()
}

Answer 1

The problem is this:

val taskList = ArrayList<Task>()

eventPlanners.forEach { eventPlanner ->

    // ...

    compositeDisposable.add(wfmStorageDomain.saveTasks(taskList)

    // ...
}

You have a list which you keep adding from different eventPlanner items and submitting that partial list to the save task. If there are more than one item in eventPlanners this will result in the ConcurrentModificationException error. Either make taskList local to the forEach loop or move the saveTasks out of forEach.