package io.quarkus.smallrye.reactivemessaging.blocking.beans;

import java.util.ArrayList;
import java.util.List;

import javax.enterprise.context.ApplicationScoped;

import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams;
import org.eclipse.microprofile.reactive.streams.operators.SubscriberBuilder;

@ApplicationScoped
public class InfiniteSubscriber {

    private final List<Integer> payloads = new ArrayList<>();
    private final List<Integer> messages = new ArrayList<>();

    @Incoming("infinite-producer-payload")
    public SubscriberBuilder<Integer, Void> consumeFourItems() {
        return ReactiveStreams
                .<Integer> builder()
                .limit(4)
                .forEach(payloads::add);
    }

    @Incoming("infinite-producer-msg")
    public SubscriberBuilder<Integer, Void> consumeFourMessages() {
        return ReactiveStreams
                .<Integer> builder()
                .limit(4)
                .forEach(messages::add);
    }

    public List<Integer> payloads() {
        return payloads;
    }

    public List<Integer> messages() {
        return messages;
    }

}

package io.quarkus.smallrye.reactivemessaging.blocking.beans;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;

import javax.enterprise.context.ApplicationScoped;

import org.eclipse.microprofile.reactive.messaging.Incoming;
import org.eclipse.microprofile.reactive.streams.operators.ReactiveStreams;
import org.eclipse.microprofile.reactive.streams.operators.SubscriberBuilder;

@ApplicationScoped
public class InfiniteSubscriber {

    private final List<Integer> payloads = new CopyOnWriteArrayList<>();
    private final List<Integer> messages = new CopyOnWriteArrayList<>();

    @Incoming("infinite-producer-payload")
    public SubscriberBuilder<Integer, Void> consumeFourItems() {
        return ReactiveStreams
                .<Integer> builder()
                .limit(4)
                .forEach(payloads::add);
    }

    @Incoming("infinite-producer-msg")
    public SubscriberBuilder<Integer, Void> consumeFourMessages() {
        return ReactiveStreams
                .<Integer> builder()
                .limit(4)
                .forEach(messages::add);
    }

    public List<Integer> payloads() {
        return payloads;
    }

    public List<Integer> messages() {
        return messages;
    }

}

Fix concurrency modification exception that could happen in the reactive messaging tests.
Fix concurrency modification exception that could happen in the reactive messaging tests.

The emission and the retrieval of the list are done on different thread. While checking the content of the retrieved list, new items can be added to the list leading to a concurrency exception.
