package com.hazelcast.map;

import com.hazelcast.core.HazelcastInstance;
import com.hazelcast.map.listener.MapPartitionLostListener;
import com.hazelcast.partition.AbstractPartitionLostListenerTest;
import com.hazelcast.test.AssertTask;
import com.hazelcast.test.HazelcastSerialClassRunner;
import com.hazelcast.test.annotation.SlowTest;
import org.junit.Test;
import org.junit.experimental.categories.Category;
import org.junit.runner.RunWith;

import java.util.ArrayList;
import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

@RunWith(HazelcastSerialClassRunner.class)
@Category(SlowTest.class)
public class MapPartitionLostListenerStressTest
        extends AbstractPartitionLostListenerTest {

    public static class EventCollectingMapPartitionLostListener
            implements MapPartitionLostListener {

        private final List<MapPartitionLostEvent> events = Collections.synchronizedList(new LinkedList<MapPartitionLostEvent>());

        private final int backupCount;

        public EventCollectingMapPartitionLostListener(int backupCount) {
            this.backupCount = backupCount;
        }

        @Override
        public void partitionLost(MapPartitionLostEvent event) {
            this.events.add(event);
        }

        public List<MapPartitionLostEvent> getEvents() {
            synchronized (events) {
                return new ArrayList<MapPartitionLostEvent>(events);
            }
        }

        public int getBackupCount() {
            return backupCount;
        }
    }

    protected int getNodeCount() {
        return 5;
    }

    protected int getMapEntryCount() {
        return 5000;
    }

    @Test
    public void test_mapPartitionLostListenerInvoked_when1NodeCrashed_withoutData()
            throws InterruptedException {
        testMapPartitionLostListener(1, false);
    }

    @Test
    public void test_mapPartitionLostListenerInvoked_when1NodeCrashed_withData()
            throws InterruptedException {
        testMapPartitionLostListener(1, true);
    }

    @Test
    public void test_mapPartitionLostListenerInvoked_when2NodesCrashed_withoutData()
            throws InterruptedException {
        testMapPartitionLostListener(2, false);
    }

    @Test
    public void test_mapPartitionLostListenerInvoked_when2NodesCrashed_withData()
            throws InterruptedException {
        testMapPartitionLostListener(2, true);
    }

    @Test
    public void test_mapPartitionLostListenerInvoked_when3NodesCrashed_withoutData()
            throws InterruptedException {
        testMapPartitionLostListener(3, false);
    }

    @Test
    public void test_mapPartitionLostListenerInvoked_when3NodesCrashed_withData()
            throws InterruptedException {
        testMapPartitionLostListener(3, true);
    }

    @Test
    public void test_mapPartitionLostListenerInvoked_when4NodesCrashed_withoutData()
            throws InterruptedException {
        testMapPartitionLostListener(4, false);
    }

    @Test
    public void test_mapPartitionLostListenerInvoked_when4NodesCrashed_withData()
            throws InterruptedException {
        testMapPartitionLostListener(4, true);
    }

    private void testMapPartitionLostListener(final int numberOfNodesToCrash, final boolean withData)
            throws InterruptedException {
        final List<HazelcastInstance> instances = getCreatedInstancesShuffledAfterWarmedUp();

        List<HazelcastInstance> survivingInstances = new ArrayList<HazelcastInstance>(instances);
        final List<HazelcastInstance> terminatingInstances = survivingInstances.subList(0, numberOfNodesToCrash);
        survivingInstances = survivingInstances.subList(numberOfNodesToCrash, instances.size());

        final List<EventCollectingMapPartitionLostListener> listeners = registerListeners(survivingInstances.get(0));

        if (withData) {
            populateMaps(survivingInstances.get(0));
        }

        final String log = "Surviving: " + survivingInstances + " Terminating: " + terminatingInstances;
        final Map<Integer, Integer> survivingPartitions = getMinReplicaIndicesByPartitionId(survivingInstances);

        terminateInstances(terminatingInstances);
        waitAllForSafeStateAndDumpPartitionServiceOnFailure(survivingInstances, 300);

        for (int i = 0; i < getNodeCount(); i++) {
            assertListenerInvocationsEventually(numberOfNodesToCrash, log, survivingPartitions, listeners.get(i), i);
        }
    }

    private void assertListenerInvocationsEventually(final int numberOfNodesToCrash, final String log,
                                                     final Map<Integer, Integer> survivingPartitions,
                                                     final EventCollectingMapPartitionLostListener listener, final int index) {
        assertTrueEventually(new AssertTask() {
            @Override
            public void run()
                    throws Exception {
                if (index < numberOfNodesToCrash) {
                    assertLostPartitions(log, listener, survivingPartitions);
                } else {
                    final String message = log + " listener-" + index + " should not be invoked!";
                    assertTrue(message, listener.getEvents().isEmpty());
                }
            }
        });
    }

    private void assertLostPartitions(final String log, final EventCollectingMapPartitionLostListener listener,
                                      final Map<Integer, Integer> survivingPartitions) {
        final List<MapPartitionLostEvent> events = listener.getEvents();

        assertFalse(survivingPartitions.isEmpty());

        for (MapPartitionLostEvent event : events) {
            final int failedPartitionId = event.getPartitionId();
            final Integer survivingReplicaIndex = survivingPartitions.get(failedPartitionId);
            if (survivingReplicaIndex != null) {
                final String message =
                        log + ", PartitionId: " + failedPartitionId + " SurvivingReplicaIndex: " + survivingReplicaIndex
                                + " Map Name: " + event.getName();
                assertTrue(message, survivingReplicaIndex > listener.getBackupCount());
            }
        }
    }

    private List<EventCollectingMapPartitionLostListener> registerListeners(final HazelcastInstance instance) {
        final List<EventCollectingMapPartitionLostListener> listeners = new ArrayList<EventCollectingMapPartitionLostListener>();
        for (int i = 0; i < getNodeCount(); i++) {
            final EventCollectingMapPartitionLostListener listener = new EventCollectingMapPartitionLostListener(i);
            instance.getMap(getIthMapName(i)).addPartitionLostListener(listener);
            listeners.add(listener);
        }

        return listeners;
    }

}

@@ -471,67 +471,8 @@
 st;%0A
-import java.util.Collections;%0Aimport java.util.LinkedList;%0A
 impo
@@ -949,47 +949,17 @@
 s = 
-Collections.synchronizedList(new Linked
+new Array
 List
@@ -983,17 +983,16 @@
 Event%3E()
-)
 ;%0A%0A     
@@ -1184,24 +1184,37 @@
      public 
+synchronized 
 void partiti
@@ -1313,16 +1313,29 @@
  public 
+synchronized 
 List%3CMap
@@ -1372,48 +1372,8 @@
 ) %7B%0A
-            synchronized (events) %7B%0A    
     
@@ -1433,30 +1433,16 @@
 vents);%0A
-            %7D%0A
         

Fix thread safety issue in map partition lost listener in test