/*
 * Copyright (c) 2008-2019, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.hazelcast.jet.impl.connector;

import com.hazelcast.core.IMap;
import com.hazelcast.jet.JetInstance;
import com.hazelcast.jet.core.JetTestSupport;
import com.hazelcast.jet.core.processor.SourceProcessors;
import com.hazelcast.jet.core.test.TestSupport;
import com.hazelcast.jet.function.FunctionEx;
import com.hazelcast.projection.Projection;
import com.hazelcast.query.Predicate;
import com.hazelcast.test.HazelcastParallelClassRunner;
import org.junit.Before;
import org.junit.Test;
import org.junit.runner.RunWith;

import java.util.ArrayList;
import java.util.List;
import java.util.Map.Entry;

import static com.hazelcast.jet.Util.entry;
import static java.util.Collections.emptyList;

@RunWith(HazelcastParallelClassRunner.class)
public class ReadMapOrCachePTest extends JetTestSupport {

    private JetInstance jet;

    @Before
    public void setUp() {
        jet = createJetMember();
    }

    @Test
    public void test_whenEmpty() {
        TestSupport
                .verifyProcessor(SourceProcessors.readMapP("map"))
                .jetInstance(jet)
                .disableSnapshots()
                .disableProgressAssertion()
                .expectOutput(emptyList());
    }

    @Test
    public void test_whenNoPredicateAndNoProjection() {
        IMap<Integer, String> map = jet.getMap("map");
        List<Entry<Integer, String>> expected = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            map.put(i, "value-" + i);
            expected.add(entry(i, "value-" + i));
        }

        TestSupport
            .verifyProcessor(SourceProcessors.readMapP("map"))
            .jetInstance(jet)
            .disableSnapshots()
            .disableProgressAssertion()
            .outputChecker(TestSupport.SAME_ITEMS_ANY_ORDER)
            .expectOutput(expected);
    }

    @Test
    public void test_whenPredicateAndProjectionSet() {
        IMap<Integer, String> map = jet.getMap("map");
        List<String> expected = new ArrayList<>();
        for (int i = 0; i < 1000; i++) {
            map.put(i, "value-" + i);
            if (i % 2 == 0) {
                expected.add("value-" + i);
            }
        }

        Predicate<Integer, String> predicate = entry -> entry.getKey() % 2 == 0;
        Projection<Entry<Integer, String>, String> projection = toProjection(Entry::getValue);
        TestSupport
            .verifyProcessor(SourceProcessors.readMapP("map", predicate, projection))
            .jetInstance(jet)
            .disableSnapshots()
            .disableProgressAssertion()
            .outputChecker(TestSupport.SAME_ITEMS_ANY_ORDER)
            .expectOutput(expected);
    }

    private static <I, O> Projection<I, O> toProjection(FunctionEx<I, O> projectionFn) {
        return new Projection<I, O>() {
            @Override public O transform(I input) {
                return projectionFn.apply(input);
            }
        };
    }
}

@@ -724,58 +724,27 @@
 jet.
-JetInstance;%0Aimport com.hazelcast.jet.core.JetTest
+SimpleTestInCluster
 Supp
@@ -1009,39 +1009,37 @@
 t.test.Hazelcast
-Paralle
+Seria
 lClassRunner;%0Aim
@@ -1059,16 +1059,21 @@
 t.Before
+Class
 ;%0Aimport
@@ -1317,15 +1317,13 @@
 cast
-Paralle
+Seria
 lCla
@@ -1383,54 +1383,36 @@
 nds 
-JetTestSupport %7B%0A%0A    private JetInstance jet;
+SimpleTestInClusterSupport %7B
 %0A%0A  
@@ -1420,16 +1420,21 @@
  @Before
+Class
 %0A    pub
@@ -1433,24 +1433,31 @@
 %0A    public 
+static 
 void setUp()
@@ -1471,30 +1471,26 @@
     
-jet = createJetMember(
+initialize(1, null
 );%0A 
@@ -1611,37 +1611,47 @@
 essors.readMapP(
-%22map%22
+randomMapName()
 ))%0A             
@@ -1658,35 +1658,42 @@
    .jetInstance(
-jet
+instance()
 )%0A              
@@ -1911,32 +1911,49 @@
 %3E map = 
-jet.getMap(%22map%22
+instance().getMap(randomMapName()
 );%0A     
@@ -2178,32 +2178,36 @@
 ort%0A            
+    
 .verifyProcessor
@@ -2233,21 +2233,29 @@
 eadMapP(
-%22
 map
-%22
+.getName()
 ))%0A     
@@ -2253,32 +2253,36 @@
 )))%0A            
+  
+  
 .jetInstance(jet
@@ -2270,37 +2270,48 @@
    .jetInstance(
-jet)%0A
+instance())%0A    
             .dis
@@ -2330,32 +2330,36 @@
 s()%0A            
+ 
+   
 .disableProgress
@@ -2362,32 +2362,36 @@
 ressAssertion()%0A
+    
             .out
@@ -2427,32 +2427,36 @@
 TEMS_ANY_ORDER)%0A
+    
             .exp
@@ -2588,24 +2588,41 @@
 p = 
-jet.getMap(%22map%22
+instance().getMap(randomMapName()
 );%0A 
@@ -3049,32 +3049,36 @@
 ort%0A            
+    
 .verifyProcessor
@@ -3108,13 +3108,21 @@
 apP(
-%22
 map
-%22
+.getName()
 , pr
@@ -3147,32 +3147,36 @@
 n))%0A            
+    
 .jetInstance(jet
@@ -3176,13 +3176,24 @@
 nce(
-jet)%0A
+instance())%0A    
     
@@ -3224,32 +3224,36 @@
 s()%0A            
+ 
+   
 .disableProgress
@@ -3256,32 +3256,36 @@
 ressAssertion()%0A
+    
             .out
@@ -3321,32 +3321,36 @@
 TEMS_ANY_ORDER)%0A
+    
             .exp

Make ReadMapOrCachePTest serial, share cluster