[BAEL-9555] - Created a core-java-modules folder

core-java-modules/core-java-concurrency-collections/.gitignore

@@ -0,0 +1,26 @@
+*.class
+
+0.*
+
+#folders#
+/target
+/neoDb*
+/data
+/src/main/webapp/WEB-INF/classes
+*/META-INF/*
+.resourceCache
+
+# Packaged files #
+*.jar
+*.war
+*.ear
+
+# Files generated by integration tests
+*.txt
+backup-pom.xml
+/bin/
+/temp
+
+#IntelliJ specific
+.idea/
+*.iml

core-java-modules/core-java-concurrency-collections/README.md

@@ -0,0 +1,15 @@
+=========
+
+## Core Java Concurrency Collections Examples
+
+### Relevant Articles: 
+- [Guide to java.util.concurrent.BlockingQueue](http://www.baeldung.com/java-blocking-queue)
+- [A Guide to ConcurrentMap](http://www.baeldung.com/java-concurrent-map)
+- [Guide to PriorityBlockingQueue in Java](http://www.baeldung.com/java-priority-blocking-queue)
+- [Avoiding the ConcurrentModificationException in Java](http://www.baeldung.com/java-concurrentmodificationexception)
+- [Custom Thread Pools In Java 8 Parallel Streams](http://www.baeldung.com/java-8-parallel-streams-custom-threadpool)
+- [Guide to DelayQueue](http://www.baeldung.com/java-delay-queue)
+- [A Guide to Java SynchronousQueue](http://www.baeldung.com/java-synchronous-queue)
+- [Guide to the Java TransferQueue](http://www.baeldung.com/java-transfer-queue)
+- [Guide to the ConcurrentSkipListMap](http://www.baeldung.com/java-concurrent-skip-list-map)
+- [Guide to CopyOnWriteArrayList](http://www.baeldung.com/java-copy-on-write-arraylist)

core-java-modules/core-java-concurrency-collections/pom.xml

@@ -0,0 +1,74 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <groupId>com.baeldung</groupId>
+    <artifactId>core-java-concurrency-collections</artifactId>
+    <version>0.1.0-SNAPSHOT</version>
+    <name>core-java-concurrency-collections</name>
+    <packaging>jar</packaging>
+
+    <parent>
+        <groupId>com.baeldung</groupId>
+        <artifactId>parent-java</artifactId>
+        <version>0.0.1-SNAPSHOT</version>
+        <relativePath>../../parent-java</relativePath>
+    </parent>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-collections4</artifactId>
+            <version>${commons-collections4.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>commons-io</groupId>
+            <artifactId>commons-io</artifactId>
+            <version>${commons-io.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>${commons-lang3.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-math3</artifactId>
+            <version>${commons-math3.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <version>${assertj.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>com.jayway.awaitility</groupId>
+            <artifactId>awaitility</artifactId>
+            <version>${avaitility.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <finalName>core-java-concurrency-collections</finalName>
+        <resources>
+            <resource>
+                <directory>src/main/resources</directory>
+                <filtering>true</filtering>
+            </resource>
+        </resources>
+    </build>
+
+    <properties>
+        <!-- util -->
+        <guava.version>21.0</guava.version>
+        <commons-lang3.version>3.5</commons-lang3.version>
+        <commons-math3.version>3.6.1</commons-math3.version>
+        <commons-collections4.version>4.1</commons-collections4.version>
+        <collections-generic.version>4.01</collections-generic.version>
+        <!-- testing -->
+        <assertj.version>3.6.1</assertj.version>
+        <avaitility.version>1.7.0</avaitility.version>
+    </properties>
+
+</project>

core-java-modules/core-java-concurrency-collections/src/main/java/com/baeldung/concurrent/blockingqueue/BlockingQueueUsage.java

@@ -0,0 +1,27 @@
+package com.baeldung.concurrent.blockingqueue;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+public class BlockingQueueUsage {
+    public static void main(String[] args) {
+        int BOUND = 10;
+        int N_PRODUCERS = 4;
+        int N_CONSUMERS = Runtime.getRuntime().availableProcessors();
+        int poisonPill = Integer.MAX_VALUE;
+        int poisonPillPerProducer = N_CONSUMERS / N_PRODUCERS;
+        int mod = N_CONSUMERS % N_PRODUCERS;
+        BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(BOUND);
+
+        for (int i = 1; i < N_PRODUCERS; i++) {
+            new Thread(new NumbersProducer(queue, poisonPill, poisonPillPerProducer)).start();
+        }
+
+        for (int j = 0; j < N_CONSUMERS; j++) {
+            new Thread(new NumbersConsumer(queue, poisonPill)).start();
+        }
+        
+        new Thread(new NumbersProducer(queue, poisonPill, poisonPillPerProducer+mod)).start();
+
+    }
+}

core-java-modules/core-java-concurrency-collections/src/main/java/com/baeldung/concurrent/blockingqueue/NumbersConsumer.java

@@ -0,0 +1,28 @@
+package com.baeldung.concurrent.blockingqueue;
+
+import java.util.concurrent.BlockingQueue;
+
+public class NumbersConsumer implements Runnable {
+    private final BlockingQueue<Integer> queue;
+    private final int poisonPill;
+
+    NumbersConsumer(BlockingQueue<Integer> queue, int poisonPill) {
+        this.queue = queue;
+        this.poisonPill = poisonPill;
+    }
+
+    public void run() {
+        try {
+            while (true) {
+                Integer number = queue.take();
+                if (number.equals(poisonPill)) {
+                    return;
+                }
+                String result = number.toString();
+                System.out.println(Thread.currentThread().getName() + " result: " + result);
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+}

core-java-modules/core-java-concurrency-collections/src/main/java/com/baeldung/concurrent/blockingqueue/NumbersProducer.java

@@ -0,0 +1,36 @@
+package com.baeldung.concurrent.blockingqueue;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ThreadLocalRandom;
+
+public class NumbersProducer implements Runnable {
+
+    private final BlockingQueue<Integer> numbersQueue;
+    private final int poisonPill;
+    private final int poisonPillPerProducer;
+
+    NumbersProducer(BlockingQueue<Integer> numbersQueue, int poisonPill, int poisonPillPerProducer) {
+        this.numbersQueue = numbersQueue;
+        this.poisonPill = poisonPill;
+        this.poisonPillPerProducer = poisonPillPerProducer;
+    }
+
+    public void run() {
+        try {
+            generateNumbers();
+        } catch (InterruptedException e) {
+            Thread.currentThread()
+                .interrupt();
+        }
+    }
+
+    private void generateNumbers() throws InterruptedException {
+        for (int i = 0; i < 100; i++) {
+            numbersQueue.put(ThreadLocalRandom.current()
+                .nextInt(100));
+        }
+        for (int j = 0; j < poisonPillPerProducer; j++) {
+            numbersQueue.put(poisonPill);
+        }
+    }
+}

core-java-modules/core-java-concurrency-collections/src/main/java/com/baeldung/concurrent/delayqueue/DelayObject.java

@@ -0,0 +1,32 @@
+package com.baeldung.concurrent.delayqueue;
+
+import com.google.common.primitives.Ints;
+
+import java.util.concurrent.Delayed;
+import java.util.concurrent.TimeUnit;
+
+public class DelayObject implements Delayed {
+    private String data;
+    private long startTime;
+
+    DelayObject(String data, long delayInMilliseconds) {
+        this.data = data;
+        this.startTime = System.currentTimeMillis() + delayInMilliseconds;
+    }
+
+    @Override
+    public long getDelay(TimeUnit unit) {
+        long diff = startTime - System.currentTimeMillis();
+        return unit.convert(diff, TimeUnit.MILLISECONDS);
+    }
+
+    @Override
+    public int compareTo(Delayed o) {
+        return Ints.saturatedCast(this.startTime - ((DelayObject) o).startTime);
+    }
+
+    @Override
+    public String toString() {
+        return "{" + "data='" + data + '\'' + ", startTime=" + startTime + '}';
+    }
+}

core-java-modules/core-java-concurrency-collections/src/main/java/com/baeldung/concurrent/delayqueue/DelayQueueConsumer.java

@@ -0,0 +1,30 @@
+package com.baeldung.concurrent.delayqueue;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+
+public class DelayQueueConsumer implements Runnable {
+    private BlockingQueue<DelayObject> queue;
+    private final Integer numberOfElementsToTake;
+    final AtomicInteger numberOfConsumedElements = new AtomicInteger();
+
+    DelayQueueConsumer(BlockingQueue<DelayObject> queue, Integer numberOfElementsToTake) {
+        this.queue = queue;
+        this.numberOfElementsToTake = numberOfElementsToTake;
+    }
+
+
+    @Override
+    public void run() {
+        for (int i = 0; i < numberOfElementsToTake; i++) {
+            try {
+                DelayObject object = queue.take();
+                numberOfConsumedElements.incrementAndGet();
+                System.out.println("Consumer take: " + object);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+}

core-java-modules/core-java-concurrency-collections/src/main/java/com/baeldung/concurrent/delayqueue/DelayQueueProducer.java

@@ -0,0 +1,34 @@
+package com.baeldung.concurrent.delayqueue;
+
+import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
+
+
+public class DelayQueueProducer implements Runnable {
+    private BlockingQueue<DelayObject> queue;
+    private final Integer numberOfElementsToProduce;
+    private final Integer delayOfEachProducedMessageMilliseconds;
+
+    DelayQueueProducer(BlockingQueue<DelayObject> queue,
+                       Integer numberOfElementsToProduce,
+                       Integer delayOfEachProducedMessageMilliseconds) {
+        this.queue = queue;
+        this.numberOfElementsToProduce = numberOfElementsToProduce;
+        this.delayOfEachProducedMessageMilliseconds = delayOfEachProducedMessageMilliseconds;
+    }
+
+    @Override
+    public void run() {
+        for (int i = 0; i < numberOfElementsToProduce; i++) {
+            DelayObject object
+                    = new DelayObject(UUID.randomUUID().toString(), delayOfEachProducedMessageMilliseconds);
+            System.out.println("Put object = " + object);
+            try {
+                queue.put(object);
+                Thread.sleep(500);
+            } catch (InterruptedException ie) {
+                ie.printStackTrace();
+            }
+        }
+    }
+}

core-java-modules/core-java-concurrency-collections/src/main/java/com/baeldung/concurrent/skiplist/Event.java

@@ -0,0 +1,21 @@
+package com.baeldung.concurrent.skiplist;
+
+import java.time.ZonedDateTime;
+
+class Event {
+    private final ZonedDateTime eventTime;
+    private final String content;
+
+    Event(ZonedDateTime eventTime, String content) {
+        this.eventTime = eventTime;
+        this.content = content;
+    }
+
+    ZonedDateTime getEventTime() {
+        return eventTime;
+    }
+
+    String getContent() {
+        return content;
+    }
+}

core-java-modules/core-java-concurrency-collections/src/main/java/com/baeldung/concurrent/skiplist/EventWindowSort.java

@@ -0,0 +1,29 @@
+package com.baeldung.concurrent.skiplist;
+
+import java.time.ZonedDateTime;
+import java.util.Comparator;
+import java.util.concurrent.ConcurrentNavigableMap;
+import java.util.concurrent.ConcurrentSkipListMap;
+
+class EventWindowSort {
+    private final ConcurrentSkipListMap<ZonedDateTime, String> events
+      = new ConcurrentSkipListMap<>(Comparator.comparingLong(value -> value.toInstant().toEpochMilli()));
+
+    void acceptEvent(Event event) {
+        events.put(event.getEventTime(), event.getContent());
+    }
+
+    ConcurrentNavigableMap<ZonedDateTime, String> getEventsFromLastMinute() {
+        return events.tailMap(ZonedDateTime
+          .now()
+          .minusMinutes(1));
+    }
+
+    ConcurrentNavigableMap<ZonedDateTime, String> getEventsOlderThatOneMinute() {
+        return events.headMap(ZonedDateTime
+          .now()
+          .minusMinutes(1));
+    }
+
+}
+

core-java-modules/core-java-concurrency-collections/src/main/java/com/baeldung/transferqueue/Consumer.java

@@ -0,0 +1,41 @@
+package com.baeldung.transferqueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.TransferQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class Consumer implements Runnable {
+    private static final Logger LOG = LoggerFactory.getLogger(Consumer.class);
+
+    private final TransferQueue<String> transferQueue;
+    private final String name;
+    final int numberOfMessagesToConsume;
+    final AtomicInteger numberOfConsumedMessages = new AtomicInteger();
+
+    Consumer(TransferQueue<String> transferQueue, String name, int numberOfMessagesToConsume) {
+        this.transferQueue = transferQueue;
+        this.name = name;
+        this.numberOfMessagesToConsume = numberOfMessagesToConsume;
+    }
+
+    @Override
+    public void run() {
+        for (int i = 0; i < numberOfMessagesToConsume; i++) {
+            try {
+                LOG.debug("Consumer: " + name + " is waiting to take element...");
+                String element = transferQueue.take();
+                longProcessing(element);
+                LOG.debug("Consumer: " + name + " received element: " + element);
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    private void longProcessing(String element) throws InterruptedException {
+        numberOfConsumedMessages.incrementAndGet();
+        Thread.sleep(500);
+    }
+}

core-java-modules/core-java-concurrency-collections/src/main/java/com/baeldung/transferqueue/Producer.java

@@ -0,0 +1,41 @@
+package com.baeldung.transferqueue;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TransferQueue;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class Producer implements Runnable {
+    private static final Logger LOG = LoggerFactory.getLogger(Producer.class);
+
+    private final TransferQueue<String> transferQueue;
+    private final String name;
+    final Integer numberOfMessagesToProduce;
+    final AtomicInteger numberOfProducedMessages = new AtomicInteger();
+
+    Producer(TransferQueue<String> transferQueue, String name, Integer numberOfMessagesToProduce) {
+        this.transferQueue = transferQueue;
+        this.name = name;
+        this.numberOfMessagesToProduce = numberOfMessagesToProduce;
+    }
+
+    @Override
+    public void run() {
+        for (int i = 0; i < numberOfMessagesToProduce; i++) {
+            try {
+                LOG.debug("Producer: " + name + " is waiting to transfer...");
+                boolean added = transferQueue.tryTransfer("A" + i, 4000, TimeUnit.MILLISECONDS);
+                if (added) {
+                    numberOfProducedMessages.incrementAndGet();
+                    LOG.debug("Producer: " + name + " transferred element: A" + i);
+                } else {
+                    LOG.debug("can not add an element due to the timeout");
+                }
+            } catch (InterruptedException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+}

core-java-modules/core-java-concurrency-collections/src/main/resources/logback.xml

@@ -0,0 +1,19 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>web - %date [%thread] %-5level %logger{36} - %message%n
+            </pattern>
+        </encoder>
+    </appender>
+
+    <logger name="org.springframework" level="WARN" />
+    <logger name="org.springframework.transaction" level="WARN" />
+
+    <!-- in order to debug some marshalling issues, this needs to be TRACE -->
+    <logger name="org.springframework.web.servlet.mvc" level="WARN" />
+
+    <root level="INFO">
+        <appender-ref ref="STDOUT" />
+    </root>
+</configuration>

core-java-modules/core-java-concurrency-collections/src/test/java/com/baeldung/concurrent/copyonwrite/CopyOnWriteArrayListUnitTest.java

@@ -0,0 +1,53 @@
+package com.baeldung.concurrent.copyonwrite;
+
+
+import org.junit.Test;
+
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+public class CopyOnWriteArrayListUnitTest {
+
+    @Test
+    public void givenCopyOnWriteList_whenIterateAndAddElementToUnderneathList_thenShouldNotChangeIterator() {
+        //given
+        final CopyOnWriteArrayList<Integer> numbers =
+          new CopyOnWriteArrayList<>(new Integer[]{1, 3, 5, 8});
+
+        //when
+        Iterator<Integer> iterator = numbers.iterator();
+        numbers.add(10);
+
+        //then
+        List<Integer> result = new LinkedList<>();
+        iterator.forEachRemaining(result::add);
+        assertThat(result).containsOnly(1, 3, 5, 8);
+
+        //and
+        Iterator<Integer> iterator2 = numbers.iterator();
+        List<Integer> result2 = new LinkedList<>();
+        iterator2.forEachRemaining(result2::add);
+
+        //then
+        assertThat(result2).containsOnly(1, 3, 5, 8, 10);
+
+    }
+
+    @Test(expected = UnsupportedOperationException.class)
+    public void givenCopyOnWriteList_whenIterateOverItAndTryToRemoveElement_thenShouldThrowException() {
+        //given
+        final CopyOnWriteArrayList<Integer> numbers =
+          new CopyOnWriteArrayList<>(new Integer[]{1, 3, 5, 8});
+
+        //when
+        Iterator<Integer> iterator = numbers.iterator();
+        while (iterator.hasNext()) {
+            iterator.remove();
+        }
+    }
+}

core-java-modules/core-java-concurrency-collections/src/test/java/com/baeldung/concurrent/delayqueue/DelayQueueIntegrationTest.java

@@ -0,0 +1,82 @@
+package com.baeldung.concurrent.delayqueue;
+
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.runners.MethodSorters;
+
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.DelayQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static junit.framework.TestCase.assertEquals;
+
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class DelayQueueIntegrationTest {
+    @Test
+    public void givenDelayQueue_whenProduceElement_thenShouldConsumeAfterGivenDelay() throws InterruptedException {
+        //given
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+        BlockingQueue<DelayObject> queue = new DelayQueue<>();
+        int numberOfElementsToProduce = 2;
+        int delayOfEachProducedMessageMilliseconds = 500;
+        DelayQueueConsumer consumer = new DelayQueueConsumer(queue, numberOfElementsToProduce);
+        DelayQueueProducer producer
+          = new DelayQueueProducer(queue, numberOfElementsToProduce, delayOfEachProducedMessageMilliseconds);
+
+        //when
+        executor.submit(producer);
+        executor.submit(consumer);
+
+        //then
+        executor.awaitTermination(5, TimeUnit.SECONDS);
+        executor.shutdown();
+        assertEquals(consumer.numberOfConsumedElements.get(), numberOfElementsToProduce);
+
+    }
+
+    @Test
+    public void givenDelayQueue_whenProduceElementWithHugeDelay_thenConsumerWasNotAbleToConsumeMessageInGivenTime() throws InterruptedException {
+        //given
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+        BlockingQueue<DelayObject> queue = new DelayQueue<>();
+        int numberOfElementsToProduce = 1;
+        int delayOfEachProducedMessageMilliseconds = 10_000;
+        DelayQueueConsumer consumer = new DelayQueueConsumer(queue, numberOfElementsToProduce);
+        DelayQueueProducer producer
+          = new DelayQueueProducer(queue, numberOfElementsToProduce, delayOfEachProducedMessageMilliseconds);
+
+        //when
+        executor.submit(producer);
+        executor.submit(consumer);
+
+        //then
+        executor.awaitTermination(5, TimeUnit.SECONDS);
+        executor.shutdown();
+        assertEquals(consumer.numberOfConsumedElements.get(), 0);
+
+    }
+
+    @Test
+    public void givenDelayQueue_whenProduceElementWithNegativeDelay_thenConsumeMessageImmediately() throws InterruptedException {
+        //given
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+        BlockingQueue<DelayObject> queue = new DelayQueue<>();
+        int numberOfElementsToProduce = 1;
+        int delayOfEachProducedMessageMilliseconds = -10_000;
+        DelayQueueConsumer consumer = new DelayQueueConsumer(queue, numberOfElementsToProduce);
+        DelayQueueProducer producer
+          = new DelayQueueProducer(queue, numberOfElementsToProduce, delayOfEachProducedMessageMilliseconds);
+
+        //when
+        executor.submit(producer);
+        executor.submit(consumer);
+
+        //then
+        executor.awaitTermination(1, TimeUnit.SECONDS);
+        executor.shutdown();
+        assertEquals(consumer.numberOfConsumedElements.get(), 1);
+
+    }
+}

core-java-modules/core-java-concurrency-collections/src/test/java/com/baeldung/concurrent/priorityblockingqueue/PriorityBlockingQueueIntegrationTest.java

@@ -0,0 +1,57 @@
+package com.baeldung.concurrent.priorityblockingqueue;
+
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.concurrent.PriorityBlockingQueue;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.util.Lists.newArrayList;
+
+public class PriorityBlockingQueueIntegrationTest {
+
+    private static final Logger LOG = LoggerFactory.getLogger(PriorityBlockingQueueIntegrationTest.class);
+
+
+    @Test
+    public void givenUnorderedValues_whenPolling_thenShouldOrderQueue() throws InterruptedException {
+        PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>();
+        ArrayList<Integer> polledElements = new ArrayList<>();
+
+        queue.add(1);
+        queue.add(5);
+        queue.add(2);
+        queue.add(3);
+        queue.add(4);
+
+        queue.drainTo(polledElements);
+
+        assertThat(polledElements).containsExactly(1, 2, 3, 4, 5);
+    }
+
+    @Test
+    public void whenPollingEmptyQueue_thenShouldBlockThread() throws InterruptedException {
+        PriorityBlockingQueue<Integer> queue = new PriorityBlockingQueue<>();
+
+        final Thread thread = new Thread(() -> {
+            LOG.debug("Polling...");
+            while (true) {
+                try {
+                    Integer poll = queue.take();
+                    LOG.debug("Polled: " + poll);
+                } catch (InterruptedException ignored) {
+                }
+            }
+        });
+        thread.start();
+
+        Thread.sleep(TimeUnit.SECONDS.toMillis(5));
+        LOG.debug("Adding to queue");
+
+        queue.addAll(newArrayList(1, 5, 6, 1, 2, 6, 7));
+        Thread.sleep(TimeUnit.SECONDS.toMillis(1));
+    }
+}

core-java-modules/core-java-concurrency-collections/src/test/java/com/baeldung/concurrent/skiplist/ConcurrentSkipListSetIntegrationTest.java

@@ -0,0 +1,120 @@
+package com.baeldung.concurrent.skiplist;
+
+import org.junit.Test;
+
+import java.time.ZonedDateTime;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentNavigableMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.IntStream;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class ConcurrentSkipListSetIntegrationTest {
+
+    @Test
+    public void givenThreadsProducingEvents_whenGetForEventsFromLastMinute_thenReturnThoseEventsInTheLockFreeWay() throws InterruptedException {
+        //given
+        ExecutorService executorService = Executors.newFixedThreadPool(3);
+        EventWindowSort eventWindowSort = new EventWindowSort();
+        int numberOfThreads = 2;
+        //when
+        Runnable producer = () -> IntStream
+          .rangeClosed(0, 100)
+          .forEach(index -> eventWindowSort.acceptEvent(new Event(ZonedDateTime
+            .now()
+            .minusSeconds(index), UUID
+            .randomUUID()
+            .toString())));
+
+        for (int i = 0; i < numberOfThreads; i++) {
+            executorService.execute(producer);
+        }
+
+        Thread.sleep(500);
+
+        ConcurrentNavigableMap<ZonedDateTime, String> eventsFromLastMinute = eventWindowSort.getEventsFromLastMinute();
+
+        long eventsOlderThanOneMinute = eventsFromLastMinute
+          .entrySet()
+          .stream()
+          .filter(e -> e
+            .getKey()
+            .isBefore(ZonedDateTime
+              .now()
+              .minusMinutes(1)))
+          .count();
+        assertEquals(eventsOlderThanOneMinute, 0);
+
+        long eventYoungerThanOneMinute = eventsFromLastMinute
+          .entrySet()
+          .stream()
+          .filter(e -> e
+            .getKey()
+            .isAfter(ZonedDateTime
+              .now()
+              .minusMinutes(1)))
+          .count();
+
+        //then
+        assertTrue(eventYoungerThanOneMinute > 0);
+
+        executorService.awaitTermination(1, TimeUnit.SECONDS);
+        executorService.shutdown();
+    }
+
+    @Test
+    public void givenThreadsProducingEvents_whenGetForEventsOlderThanOneMinute_thenReturnThoseEventsInTheLockFreeWay() throws InterruptedException {
+        //given
+        ExecutorService executorService = Executors.newFixedThreadPool(3);
+        EventWindowSort eventWindowSort = new EventWindowSort();
+        int numberOfThreads = 2;
+        //when
+        Runnable producer = () -> IntStream
+          .rangeClosed(0, 100)
+          .forEach(index -> eventWindowSort.acceptEvent(new Event(ZonedDateTime
+            .now()
+            .minusSeconds(index), UUID
+            .randomUUID()
+            .toString())));
+
+        for (int i = 0; i < numberOfThreads; i++) {
+            executorService.execute(producer);
+        }
+
+        Thread.sleep(500);
+
+        ConcurrentNavigableMap<ZonedDateTime, String> eventsFromLastMinute = eventWindowSort.getEventsOlderThatOneMinute();
+
+        long eventsOlderThanOneMinute = eventsFromLastMinute
+          .entrySet()
+          .stream()
+          .filter(e -> e
+            .getKey()
+            .isBefore(ZonedDateTime
+              .now()
+              .minusMinutes(1)))
+          .count();
+        assertTrue(eventsOlderThanOneMinute > 0);
+
+        long eventYoungerThanOneMinute = eventsFromLastMinute
+          .entrySet()
+          .stream()
+          .filter(e -> e
+            .getKey()
+            .isAfter(ZonedDateTime
+              .now()
+              .minusMinutes(1)))
+          .count();
+
+        //then
+        assertEquals(eventYoungerThanOneMinute, 0);
+
+        executorService.awaitTermination(1, TimeUnit.SECONDS);
+        executorService.shutdown();
+    }
+
+}

core-java-modules/core-java-concurrency-collections/src/test/java/com/baeldung/java/concurrentmap/ConcurrentMapAggregateStatusManualTest.java

@@ -0,0 +1,78 @@
+package com.baeldung.java.concurrentmap;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+
+public class ConcurrentMapAggregateStatusManualTest {
+
+    private ExecutorService executorService;
+    private Map<String, Integer> concurrentMap;
+    private List<Integer> mapSizes;
+    private int MAX_SIZE = 100000;
+
+    @Before
+    public void init() {
+        executorService = Executors.newFixedThreadPool(2);
+        concurrentMap = new ConcurrentHashMap<>();
+        mapSizes = new ArrayList<>(MAX_SIZE);
+    }
+
+    @Test
+    public void givenConcurrentMap_whenSizeWithoutConcurrentUpdates_thenCorrect() throws InterruptedException {
+        Runnable collectMapSizes = () -> {
+            for (int i = 0; i < MAX_SIZE; i++) {
+                concurrentMap.put(String.valueOf(i), i);
+                mapSizes.add(concurrentMap.size());
+            }
+        };
+        Runnable retrieveMapData = () -> {
+            for (int i = 0; i < MAX_SIZE; i++) {
+                concurrentMap.get(String.valueOf(i));
+            }
+        };
+        executorService.execute(retrieveMapData);
+        executorService.execute(collectMapSizes);
+        executorService.shutdown();
+        executorService.awaitTermination(1, TimeUnit.MINUTES);
+
+        for (int i = 1; i <= MAX_SIZE; i++) {
+            assertEquals("map size should be consistently reliable", i, mapSizes.get(i - 1)
+                .intValue());
+        }
+        assertEquals(MAX_SIZE, concurrentMap.size());
+    }
+
+    @Test
+    public void givenConcurrentMap_whenUpdatingAndGetSize_thenError() throws InterruptedException {
+        Runnable collectMapSizes = () -> {
+            for (int i = 0; i < MAX_SIZE; i++) {
+                mapSizes.add(concurrentMap.size());
+            }
+        };
+        Runnable updateMapData = () -> {
+            for (int i = 0; i < MAX_SIZE; i++) {
+                concurrentMap.put(String.valueOf(i), i);
+            }
+        };
+        executorService.execute(updateMapData);
+        executorService.execute(collectMapSizes);
+        executorService.shutdown();
+        executorService.awaitTermination(1, TimeUnit.MINUTES);
+
+        assertNotEquals("map size collected with concurrent updates not reliable", MAX_SIZE, mapSizes.get(MAX_SIZE - 1)
+            .intValue());
+        assertEquals(MAX_SIZE, concurrentMap.size());
+    }
+
+}

core-java-modules/core-java-concurrency-collections/src/test/java/com/baeldung/java/concurrentmap/ConcurrentMapNullKeyValueManualTest.java

@@ -0,0 +1,160 @@
+package com.baeldung.java.concurrentmap;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+import static org.junit.Assert.assertNull;
+
+public class ConcurrentMapNullKeyValueManualTest {
+
+    private ConcurrentMap<String, Object> concurrentMap;
+
+    @Before
+    public void setup() {
+        concurrentMap = new ConcurrentHashMap<>();
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenGetWithNullKey_thenThrowsNPE() {
+        concurrentMap.get(null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenGetOrDefaultWithNullKey_thenThrowsNPE() {
+        concurrentMap.getOrDefault(null, new Object());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenPutWithNullKey_thenThrowsNPE() {
+        concurrentMap.put(null, new Object());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenPutNullValue_thenThrowsNPE() {
+        concurrentMap.put("test", null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMapAndKeyAbsent_whenPutWithNullKey_thenThrowsNPE() {
+        concurrentMap.putIfAbsent(null, new Object());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMapAndMapWithNullKey_whenPutNullKeyMap_thenThrowsNPE() {
+        Map<String, Object> nullKeyMap = new HashMap<>();
+        nullKeyMap.put(null, new Object());
+        concurrentMap.putAll(nullKeyMap);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMapAndMapWithNullValue_whenPutNullValueMap_thenThrowsNPE() {
+        Map<String, Object> nullValueMap = new HashMap<>();
+        nullValueMap.put("test", null);
+        concurrentMap.putAll(nullValueMap);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenReplaceNullKeyWithValues_thenThrowsNPE() {
+        concurrentMap.replace(null, new Object(), new Object());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenReplaceWithNullNewValue_thenThrowsNPE() {
+        Object o = new Object();
+        concurrentMap.replace("test", o, null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenReplaceOldNullValue_thenThrowsNPE() {
+        Object o = new Object();
+        concurrentMap.replace("test", null, o);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenReplaceWithNullValue_thenThrowsNPE() {
+        concurrentMap.replace("test", null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenReplaceNullKey_thenThrowsNPE() {
+        concurrentMap.replace(null, "test");
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenReplaceAllMappingNull_thenThrowsNPE() {
+        concurrentMap.put("test", new Object());
+        concurrentMap.replaceAll((s, o) -> null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenRemoveNullKey_thenThrowsNPE() {
+        concurrentMap.remove(null);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenRemoveNullKeyWithValue_thenThrowsNPE() {
+        concurrentMap.remove(null, new Object());
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenMergeNullKeyWithValue_thenThrowsNPE() {
+        concurrentMap.merge(null, new Object(), (o, o2) -> o2);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenMergeKeyWithNullValue_thenThrowsNPE() {
+        concurrentMap.put("test", new Object());
+        concurrentMap.merge("test", null, (o, o2) -> o2);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMapAndAssumeKeyAbsent_whenComputeWithNullKey_thenThrowsNPE() {
+        concurrentMap.computeIfAbsent(null, s -> s);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMapAndAssumeKeyPresent_whenComputeWithNullKey_thenThrowsNPE() {
+        concurrentMap.computeIfPresent(null, (s, o) -> o);
+    }
+
+    @Test(expected = NullPointerException.class)
+    public void givenConcurrentHashMap_whenComputeWithNullKey_thenThrowsNPE() {
+        concurrentMap.compute(null, (s, o) -> o);
+    }
+
+    @Test
+    public void givenConcurrentHashMap_whenMergeKeyRemappingNull_thenRemovesMapping() {
+        Object oldValue = new Object();
+        concurrentMap.put("test", oldValue);
+        concurrentMap.merge("test", new Object(), (o, o2) -> null);
+        assertNull(concurrentMap.get("test"));
+    }
+
+    @Test
+    public void givenConcurrentHashMapAndKeyAbsent_whenComputeWithKeyRemappingNull_thenRemainsAbsent() {
+        concurrentMap.computeIfPresent("test", (s, o) -> null);
+        assertNull(concurrentMap.get("test"));
+    }
+
+    @Test
+    public void givenKeyPresent_whenComputeIfPresentRemappingNull_thenMappingRemoved() {
+        Object oldValue = new Object();
+        concurrentMap.put("test", oldValue);
+        concurrentMap.computeIfPresent("test", (s, o) -> null);
+        assertNull(concurrentMap.get("test"));
+    }
+
+    @Test
+    public void givenKeyPresent_whenComputeRemappingNull_thenMappingRemoved() {
+        Object oldValue = new Object();
+        concurrentMap.put("test", oldValue);
+        concurrentMap.compute("test", (s, o) -> null);
+        assertNull(concurrentMap.get("test"));
+    }
+
+}

core-java-modules/core-java-concurrency-collections/src/test/java/com/baeldung/java/concurrentmap/ConcurrentMapPerformanceManualTest.java

@@ -0,0 +1,99 @@
+package com.baeldung.java.concurrentmap;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Hashtable;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.Test;
+
+public class ConcurrentMapPerformanceManualTest {
+
+    @Test
+    public void givenMaps_whenGetPut500KTimes_thenConcurrentMapFaster() throws Exception {
+        final Map<String, Object> hashtable = new Hashtable<>();
+        final Map<String, Object> synchronizedHashMap = Collections.synchronizedMap(new HashMap<>());
+        final Map<String, Object> concurrentHashMap = new ConcurrentHashMap<>();
+
+        final long hashtableAvgRuntime = timeElapseForGetPut(hashtable);
+        final long syncHashMapAvgRuntime = timeElapseForGetPut(synchronizedHashMap);
+        final long concurrentHashMapAvgRuntime = timeElapseForGetPut(concurrentHashMap);
+
+        System.out.println(String.format("Hashtable: %s, syncHashMap: %s, ConcurrentHashMap: %s", hashtableAvgRuntime, syncHashMapAvgRuntime, concurrentHashMapAvgRuntime));
+
+        assertTrue(hashtableAvgRuntime > concurrentHashMapAvgRuntime);
+        assertTrue(syncHashMapAvgRuntime > concurrentHashMapAvgRuntime);
+
+    }
+
+    private long timeElapseForGetPut(Map<String, Object> map) throws InterruptedException {
+        final ExecutorService executorService = Executors.newFixedThreadPool(4);
+        final long startTime = System.nanoTime();
+        for (int i = 0; i < 4; i++) {
+            executorService.execute(() -> {
+                for (int j = 0; j < 500_000; j++) {
+                    final int value = ThreadLocalRandom.current().nextInt(10000);
+                    final String key = String.valueOf(value);
+                    map.put(key, value);
+                    map.get(key);
+                }
+            });
+        }
+        executorService.shutdown();
+        executorService.awaitTermination(1, TimeUnit.MINUTES);
+        return (System.nanoTime() - startTime) / 500_000;
+    }
+
+    @Test
+    public void givenConcurrentMap_whenKeyWithSameHashCode_thenPerformanceDegrades() throws InterruptedException {
+        class SameHash {
+            @Override
+            public int hashCode() {
+                return 1;
+            }
+        }
+        final int executeTimes = 5000;
+
+        final Map<SameHash, Integer> mapOfSameHash = new ConcurrentHashMap<>();
+        ExecutorService executorService = Executors.newFixedThreadPool(2);
+        final long sameHashStartTime = System.currentTimeMillis();
+        for (int i = 0; i < 2; i++) {
+            executorService.execute(() -> {
+                for (int j = 0; j < executeTimes; j++) {
+                    mapOfSameHash.put(new SameHash(), 1);
+                }
+            });
+        }
+        executorService.shutdown();
+        executorService.awaitTermination(5, TimeUnit.SECONDS);
+
+        final long mapOfSameHashDuration = System.currentTimeMillis() - sameHashStartTime;
+        final Map<Object, Integer> mapOfDefaultHash = new ConcurrentHashMap<>();
+        executorService = Executors.newFixedThreadPool(2);
+        final long defaultHashStartTime = System.currentTimeMillis();
+        for (int i = 0; i < 2; i++) {
+            executorService.execute(() -> {
+                for (int j = 0; j < executeTimes; j++) {
+                    mapOfDefaultHash.put(new Object(), 1);
+                }
+            });
+        }
+        executorService.shutdown();
+        executorService.awaitTermination(5, TimeUnit.SECONDS);
+
+        final long mapOfDefaultHashDuration = System.currentTimeMillis() - defaultHashStartTime;
+        assertEquals(executeTimes * 2, mapOfDefaultHash.size());
+        assertEquals(executeTimes * 2, mapOfSameHash.size());
+        System.out.println(String.format("same-hash: %s, default-hash: %s", mapOfSameHashDuration, mapOfDefaultHashDuration));
+        assertTrue("same hashCode() should greatly degrade performance", mapOfSameHashDuration > (mapOfDefaultHashDuration * 10));
+    }
+
+}

core-java-modules/core-java-concurrency-collections/src/test/java/com/baeldung/java/concurrentmap/ConcurrentNavigableMapManualTest.java

@@ -0,0 +1,79 @@
+package com.baeldung.java.concurrentmap;
+
+import org.junit.Test;
+
+import java.util.Iterator;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotEquals;
+import static org.junit.Assert.assertTrue;
+
+public class ConcurrentNavigableMapManualTest {
+
+    @Test
+    public void givenSkipListMap_whenAccessInMultiThreads_thenOrderingStable() throws InterruptedException {
+        NavigableMap<Integer, String> skipListMap = new ConcurrentSkipListMap<>();
+
+        updateMapConcurrently(skipListMap);
+
+        Iterator<Integer> skipListIter = skipListMap.keySet().iterator();
+        int previous = skipListIter.next();
+        while (skipListIter.hasNext()) {
+            int current = skipListIter.next();
+            assertTrue(previous < current);
+        }
+    }
+
+    private void updateMapConcurrently(NavigableMap<Integer, String> navigableMap) throws InterruptedException {
+        ExecutorService executorService = Executors.newFixedThreadPool(4);
+        for (int i = 0; i < 4; i++) {
+            executorService.execute(() -> {
+                ThreadLocalRandom random = ThreadLocalRandom.current();
+                for (int j = 0; j < 10000; j++) {
+                    navigableMap.put(random.nextInt(), "test");
+                }
+            });
+        }
+        executorService.shutdown();
+        executorService.awaitTermination(1, TimeUnit.MINUTES);
+    }
+
+    @Test
+    public void givenSkipListMap_whenNavConcurrently_thenCountCorrect() throws InterruptedException {
+        NavigableMap<Integer, Integer> skipListMap = new ConcurrentSkipListMap<>();
+        int count = countMapElementByPollingFirstEntry(skipListMap);
+        assertEquals(10000 * 4, count);
+    }
+
+    @Test
+    public void givenTreeMap_whenNavConcurrently_thenCountError() throws InterruptedException {
+        NavigableMap<Integer, Integer> treeMap = new TreeMap<>();
+        int count = countMapElementByPollingFirstEntry(treeMap);
+        assertNotEquals(10000 * 4, count);
+    }
+
+    private int countMapElementByPollingFirstEntry(NavigableMap<Integer, Integer> navigableMap) throws InterruptedException {
+        for (int i = 0; i < 10000 * 4; i++) {
+            navigableMap.put(i, i);
+        }
+        AtomicInteger counter = new AtomicInteger(0);
+        ExecutorService executorService = Executors.newFixedThreadPool(4);
+        for (int j = 0; j < 4; j++) {
+            executorService.execute(() -> {
+                for (int i = 0; i < 10000; i++) {
+                    if (navigableMap.pollFirstEntry() != null) {
+                        counter.incrementAndGet();
+                    }
+                }
+            });
+        }
+        executorService.shutdown();
+        executorService.awaitTermination(1, TimeUnit.MINUTES);
+        return counter.get();
+    }
+
+}

core-java-modules/core-java-concurrency-collections/src/test/java/com/baeldung/java/concurrentmap/ConcurretMapMemoryConsistencyManualTest.java

@@ -0,0 +1,72 @@
+package com.baeldung.java.concurrentmap;
+
+import org.junit.Test;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.Assert.*;
+
+public class ConcurretMapMemoryConsistencyManualTest {
+
+    @Test
+    public void givenConcurrentMap_whenSumParallel_thenCorrect() throws Exception {
+        Map<String, Integer> map = new ConcurrentHashMap<>();
+        List<Integer> sumList = parallelSum100(map, 1000);
+        assertEquals(1, sumList.stream()
+            .distinct()
+            .count());
+        long wrongResultCount = sumList.stream()
+            .filter(num -> num != 100)
+            .count();
+        assertEquals(0, wrongResultCount);
+    }
+
+    @Test
+    public void givenHashtable_whenSumParallel_thenCorrect() throws Exception {
+        Map<String, Integer> map = new Hashtable<>();
+        List<Integer> sumList = parallelSum100(map, 1000);
+        assertEquals(1, sumList.stream()
+            .distinct()
+            .count());
+        long wrongResultCount = sumList.stream()
+            .filter(num -> num != 100)
+            .count();
+        assertEquals(0, wrongResultCount);
+    }
+
+    @Test
+    public void givenHashMap_whenSumParallel_thenError() throws Exception {
+        Map<String, Integer> map = new HashMap<>();
+        List<Integer> sumList = parallelSum100(map, 100);
+        assertNotEquals(1, sumList.stream()
+            .distinct()
+            .count());
+        long wrongResultCount = sumList.stream()
+            .filter(num -> num != 100)
+            .count();
+        assertTrue(wrongResultCount > 0);
+    }
+
+    private List<Integer> parallelSum100(Map<String, Integer> map, int executionTimes) throws InterruptedException {
+        List<Integer> sumList = new ArrayList<>(1000);
+        for (int i = 0; i < executionTimes; i++) {
+            map.put("test", 0);
+            ExecutorService executorService = Executors.newFixedThreadPool(4);
+            for (int j = 0; j < 10; j++) {
+                executorService.execute(() -> {
+                    for (int k = 0; k < 10; k++)
+                        map.computeIfPresent("test", (key, value) -> value + 1);
+                });
+            }
+            executorService.shutdown();
+            executorService.awaitTermination(5, TimeUnit.SECONDS);
+            sumList.add(map.get("test"));
+        }
+        return sumList;
+    }
+
+}

core-java-modules/core-java-concurrency-collections/src/test/java/com/baeldung/java/concurrentmodification/ConcurrentModificationUnitTest.java

@@ -0,0 +1,79 @@
+package com.baeldung.java.concurrentmodification;
+
+import org.junit.Test;
+
+import java.util.Collection;
+import java.util.ConcurrentModificationException;
+import java.util.Iterator;
+import java.util.List;
+
+import static java.util.stream.Collectors.toList;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.util.Lists.newArrayList;
+
+public class ConcurrentModificationUnitTest {
+    @Test(expected = ConcurrentModificationException.class)
+    public void givenIterating_whenRemoving_thenThrowException() throws InterruptedException {
+
+        List<Integer> integers = newArrayList(1, 2, 3);
+
+        for (Integer integer : integers) {
+            integers.remove(1);
+        }
+    }
+
+    @Test
+    public void givenIterating_whenUsingIteratorRemove_thenNoError() throws InterruptedException {
+
+        List<Integer> integers = newArrayList(1, 2, 3);
+
+        for (Iterator<Integer> iterator = integers.iterator(); iterator.hasNext(); ) {
+            Integer integer = iterator.next();
+            if (integer == 2) {
+                iterator.remove();
+            }
+        }
+
+        assertThat(integers).containsExactly(1, 3);
+    }
+
+    @Test
+    public void givenIterating_whenUsingRemovalList_thenNoError() throws InterruptedException {
+
+        List<Integer> integers = newArrayList(1, 2, 3);
+        List<Integer> toRemove = newArrayList();
+
+        for (Integer integer : integers) {
+            if (integer == 2) {
+                toRemove.add(integer);
+            }
+        }
+        integers.removeAll(toRemove);
+
+        assertThat(integers).containsExactly(1, 3);
+    }
+
+    @Test
+    public void whenUsingRemoveIf_thenRemoveElements() throws InterruptedException {
+
+        Collection<Integer> integers = newArrayList(1, 2, 3);
+
+        integers.removeIf(i -> i == 2);
+
+        assertThat(integers).containsExactly(1, 3);
+    }
+
+    @Test
+    public void whenUsingStream_thenRemoveElements() {
+        Collection<Integer> integers = newArrayList(1, 2, 3);
+
+        List<String> collected = integers
+          .stream()
+          .filter(i -> i != 2)
+          .map(Object::toString)
+          .collect(toList());
+
+        assertThat(collected).containsExactly("1", "3");
+    }
+
+}

core-java-modules/core-java-concurrency-collections/src/test/java/com/baeldung/synchronousqueue/SynchronousQueueIntegrationTest.java

@@ -0,0 +1,88 @@
+package com.baeldung.synchronousqueue;
+
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.runners.MethodSorters;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static junit.framework.TestCase.assertEquals;
+
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class SynchronousQueueIntegrationTest {
+
+    private static final Logger LOG = LoggerFactory.getLogger(SynchronousQueueIntegrationTest.class);
+
+
+    @Test
+    public void givenTwoThreads_whenWantToExchangeUsingLockGuardedVariable_thenItSucceed() throws InterruptedException {
+        //given
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+        AtomicInteger sharedState = new AtomicInteger();
+        CountDownLatch countDownLatch = new CountDownLatch(1);
+
+        Runnable producer = () -> {
+            Integer producedElement = ThreadLocalRandom.current().nextInt();
+            LOG.debug("Saving an element: " + producedElement + " to the exchange point");
+            sharedState.set(producedElement);
+            countDownLatch.countDown();
+        };
+
+        Runnable consumer = () -> {
+            try {
+                countDownLatch.await();
+                Integer consumedElement = sharedState.get();
+                LOG.debug("consumed an element: " + consumedElement + " from the exchange point");
+            } catch (InterruptedException ex) {
+                ex.printStackTrace();
+            }
+        };
+
+        //when
+        executor.execute(producer);
+        executor.execute(consumer);
+
+        //then
+        executor.awaitTermination(500, TimeUnit.MILLISECONDS);
+        executor.shutdown();
+        assertEquals(countDownLatch.getCount(), 0);
+    }
+
+    @Test
+    public void givenTwoThreads_whenWantToExchangeUsingSynchronousQueue_thenItSucceed() throws InterruptedException {
+        //given
+        ExecutorService executor = Executors.newFixedThreadPool(2);
+        final SynchronousQueue<Integer> queue = new SynchronousQueue<>();
+
+        Runnable producer = () -> {
+            Integer producedElement = ThreadLocalRandom.current().nextInt();
+            try {
+                LOG.debug("Saving an element: " + producedElement + " to the exchange point");
+                queue.put(producedElement);
+            } catch (InterruptedException ex) {
+                ex.printStackTrace();
+            }
+        };
+
+        Runnable consumer = () -> {
+            try {
+                Integer consumedElement = queue.take();
+                LOG.debug("consumed an element: " + consumedElement + " from the exchange point");
+            } catch (InterruptedException ex) {
+                ex.printStackTrace();
+            }
+        };
+
+        //when
+        executor.execute(producer);
+        executor.execute(consumer);
+
+        //then
+        executor.awaitTermination(500, TimeUnit.MILLISECONDS);
+        executor.shutdown();
+        assertEquals(queue.size(), 0);
+    }
+}

core-java-modules/core-java-concurrency-collections/src/test/java/com/baeldung/transferqueue/TransferQueueIntegrationTest.java

@@ -0,0 +1,74 @@
+package com.baeldung.transferqueue;
+
+import org.junit.FixMethodOrder;
+import org.junit.Test;
+import org.junit.runners.MethodSorters;
+
+import java.util.concurrent.*;
+
+import static junit.framework.TestCase.assertEquals;
+
+@FixMethodOrder(MethodSorters.NAME_ASCENDING)
+public class TransferQueueIntegrationTest {
+
+    @Test
+    public void whenMultipleConsumersAndProducers_thenProcessAllMessages() throws InterruptedException {
+        //given
+        TransferQueue<String> transferQueue = new LinkedTransferQueue<>();
+        ExecutorService exService = Executors.newFixedThreadPool(3);
+        Producer producer1 = new Producer(transferQueue, "1", 3);
+        Producer producer2 = new Producer(transferQueue, "2", 3);
+        Consumer consumer1 = new Consumer(transferQueue, "1", 3);
+        Consumer consumer2 = new Consumer(transferQueue, "2", 3);
+
+        //when
+        exService.execute(producer1);
+        exService.execute(producer2);
+        exService.execute(consumer1);
+        exService.execute(consumer2);
+
+        //then
+        exService.awaitTermination(5000, TimeUnit.MILLISECONDS);
+        exService.shutdown();
+
+        assertEquals(producer1.numberOfProducedMessages.intValue(), 3);
+        assertEquals(producer2.numberOfProducedMessages.intValue(), 3);
+    }
+
+    @Test
+    public void whenUseOneConsumerAndOneProducer_thenShouldProcessAllMessages() throws InterruptedException {
+        //given
+        TransferQueue<String> transferQueue = new LinkedTransferQueue<>();
+        ExecutorService exService = Executors.newFixedThreadPool(2);
+        Producer producer = new Producer(transferQueue, "1", 3);
+        Consumer consumer = new Consumer(transferQueue, "1", 3);
+
+        //when
+        exService.execute(producer);
+        exService.execute(consumer);
+
+        //then
+        exService.awaitTermination(5000, TimeUnit.MILLISECONDS);
+        exService.shutdown();
+
+        assertEquals(producer.numberOfProducedMessages.intValue(), 3);
+        assertEquals(consumer.numberOfConsumedMessages.intValue(), 3);
+    }
+
+    @Test
+    public void whenUseOneProducerAndNoConsumers_thenShouldFailWithTimeout() throws InterruptedException {
+        //given
+        TransferQueue<String> transferQueue = new LinkedTransferQueue<>();
+        ExecutorService exService = Executors.newFixedThreadPool(2);
+        Producer producer = new Producer(transferQueue, "1", 3);
+
+        //when
+        exService.execute(producer);
+
+        //then
+        exService.awaitTermination(5000, TimeUnit.MILLISECONDS);
+        exService.shutdown();
+
+        assertEquals(producer.numberOfProducedMessages.intValue(), 0);
+    }
+}

core-java-modules/core-java-concurrency-collections/src/test/java/org/baeldung/java/streams/ThreadPoolInParallelStreamIntegrationTest.java

@@ -0,0 +1,41 @@
+package org.baeldung.java.streams;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ForkJoinPool;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
+import java.util.stream.Stream;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class ThreadPoolInParallelStreamIntegrationTest {
+
+    @Test
+    public void giveRangeOfLongs_whenSummedInParallel_shouldBeEqualToExpectedTotal() throws InterruptedException, ExecutionException {
+        long firstNum = 1;
+        long lastNum = 1_000_000;
+
+        List<Long> aList = LongStream.rangeClosed(firstNum, lastNum).boxed().collect(Collectors.toList());
+
+        ForkJoinPool customThreadPool = new ForkJoinPool(4);
+        long actualTotal = customThreadPool
+          .submit(() -> aList.parallelStream()
+            .reduce(0L, Long::sum))
+          .get();
+
+        assertEquals((lastNum + firstNum) * lastNum / 2, actualTotal);
+    }
+
+    @Test
+    public void givenList_whenCallingParallelStream_shouldBeParallelStream() {
+        List<Long> aList = new ArrayList<>();
+        Stream<Long> parallelStream = aList.parallelStream();
+
+        assertTrue(parallelStream.isParallel());
+    }
+}

core-java-modules/core-java-concurrency-collections/src/test/resources/.gitignore

@@ -0,0 +1,13 @@
+*.class
+
+#folders#
+/target
+/neoDb*
+/data
+/src/main/webapp/WEB-INF/classes
+*/META-INF/*
+
+# Packaged files #
+*.jar
+*.war
+*.ear