CountKeyVal extends BaseKeyValueOperator<K, V> { @SuppressWarnings({ "unchecked", "rawtypes" }) @Override public void endWindow() { for (Map.Entry<K, MutableInt> e : counts.entrySet()) { count.emit(new KeyValPair(e.getKey(), new Integer(e.getValue().intValue()))); } counts.clear(); }  @SuppressWarnings({ "unchecked", "rawtypes" }) @Override void endWindow(); final transient DefaultInputPort<KeyValPair<K, V>> data; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<KeyValPair<K, Integer>> count; }
@SuppressWarnings({ "rawtypes", "unchecked" }) @Test public void testNodeProcessing() { CountKeyVal<String, Double> oper = new CountKeyVal<String, Double>(); CollectorTestSink countSink = new CollectorTestSink(); oper.count.setSink(countSink); oper.beginWindow(0); oper.data.process(new KeyValPair("a", 2.0)); oper.data.process(new KeyValPair("b", 20.0)); oper.data.process(new KeyValPair("c", 1000.0)); oper.data.process(new KeyValPair("a", 1.0)); oper.data.process(new KeyValPair("a", 10.0)); oper.data.process(new KeyValPair("b", 5.0)); oper.data.process(new KeyValPair("d", 55.0)); oper.data.process(new KeyValPair("b", 12.0)); oper.data.process(new KeyValPair("d", 22.0)); oper.data.process(new KeyValPair("d", 14.2)); oper.data.process(new KeyValPair("d", 46.0)); oper.data.process(new KeyValPair("e", 2.0)); oper.data.process(new KeyValPair("a", 23.0)); oper.data.process(new KeyValPair("d", 4.0)); oper.endWindow(); Assert.assertEquals("number emitted tuples", 5, countSink.collectedTuples.size()); for (Object o : countSink.collectedTuples) { KeyValPair<String, Integer> e = (KeyValPair<String, Integer>)o; Integer val = (Integer)e.getValue(); if (e.getKey().equals("a")) { Assert.assertEquals("emitted value for 'a' was ", 4, val.intValue()); } else if (e.getKey().equals("b")) { Assert.assertEquals("emitted tuple for 'b' was ", 3, val.intValue()); } else if (e.getKey().equals("c")) { Assert.assertEquals("emitted tuple for 'c' was ", 1, val.intValue()); } else if (e.getKey().equals("d")) { Assert.assertEquals("emitted tuple for 'd' was ", 5, val.intValue()); } else if (e.getKey().equals("e")) { Assert.assertEquals("emitted tuple for 'e' was ", 1, val.intValue()); } } }