RoundRobinHashMap extends BaseKeyValueOperator<K, V> { public void setKeys(K[] keys) { this.keys = keys; }  void setKeys(K[] keys); final transient DefaultInputPort<V> data; final transient DefaultOutputPort<HashMap<K, V>> map; }
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testNodeProcessing() throws Exception { RoundRobinHashMap oper = new RoundRobinHashMap(); CollectorTestSink mapSink = new CollectorTestSink(); String[] keys = new String[3]; keys[0] = "a"; keys[1] = "b"; keys[2] = "c"; oper.setKeys(keys); oper.map.setSink(mapSink); oper.beginWindow(0); HashMap<String, Integer> t1 = new HashMap<String, Integer>(); t1.put("a", 0); t1.put("b", 1); t1.put("c", 2); HashMap<String, Integer> t2 = new HashMap<String, Integer>(); t2.put("a", 3); t2.put("b", 4); t2.put("c", 5); HashMap<String, Integer> t3 = new HashMap<String, Integer>(); t3.put("a", 6); t3.put("b", 7); t3.put("c", 8); HashMap<String, Integer> t4 = new HashMap<String, Integer>(); t4.put("a", 9); t4.put("b", 10); t4.put("c", 11); int numtuples = 12; for (int i = 0; i < numtuples; i++) { oper.data.process(i); } oper.endWindow(); Assert.assertEquals("number emitted tuples", numtuples / 3, mapSink.collectedTuples.size()); log.debug(mapSink.collectedTuples.toString()); Assert.assertEquals("tuple 1", t1, mapSink.collectedTuples.get(0)); Assert.assertEquals("tuple 2", t2, mapSink.collectedTuples.get(1)); Assert.assertEquals("tuple 3", t3, mapSink.collectedTuples.get(2)); Assert.assertEquals("tuple 4", t4, mapSink.collectedTuples.get(3)); }