SumKeyVal extends BaseNumberKeyValueOperator<K, V> { @Override public void endWindow() { for (Map.Entry<K, SumEntry> e: sums.entrySet()) { K key = e.getKey(); SumEntry val = e.getValue(); if (val.changed) { sum.emit(new KeyValPair<K, V>(key, getValue(val.sum.doubleValue()))); sumDouble.emit(new KeyValPair<K, Double>(key, val.sum.doubleValue())); sumInteger.emit(new KeyValPair<K, Integer>(key, val.sum.intValue())); sumFloat.emit(new KeyValPair<K, Float>(key, val.sum.floatValue())); sumShort.emit(new KeyValPair<K, Short>(key, val.sum.shortValue())); sumLong.emit(new KeyValPair<K, Long>(key, val.sum.longValue())); } } clearCache(); }  boolean isCumulative(); void setCumulative(boolean cumulative); @Override void endWindow(); void clearCache(); final transient DefaultInputPort<KeyValPair<K, V>> data; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<KeyValPair<K, V>> sum; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<KeyValPair<K, Double>> sumDouble; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<KeyValPair<K, Integer>> sumInteger; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<KeyValPair<K, Long>> sumLong; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<KeyValPair<K, Short>> sumShort; @OutputPortFieldAnnotation(optional = true)
  final transient DefaultOutputPort<KeyValPair<K, Float>> sumFloat; }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void testNodeProcessing() { SumKeyVal<String, Double> oper = new SumKeyVal<String, Double>(); oper.setType(Double.class); CollectorTestSink sumSink = new CollectorTestSink(); oper.sum.setSink(sumSink); 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, sumSink.collectedTuples.size()); for (Object o: sumSink.collectedTuples) { KeyValPair<String, Double> e = (KeyValPair<String, Double>)o; Double val = (Double)e.getValue(); if (e.getKey().equals("a")) { Assert.assertEquals("emitted value for 'a' was ", new Double(36), val); } else if (e.getKey().equals("b")) { Assert.assertEquals("emitted tuple for 'b' was ", new Double(37), val); } else if (e.getKey().equals("c")) { Assert.assertEquals("emitted tuple for 'c' was ", new Double(1000), val); } else if (e.getKey().equals("d")) { Assert.assertEquals("emitted tuple for 'd' was ", new Double(141.2), val); } else if (e.getKey().equals("e")) { Assert.assertEquals("emitted tuple for 'e' was ", new Double(2), val); } } }