@SuppressWarnings({ "rawtypes", "unchecked" })

@Test
public void testNodeProcessing()
{
  FilteredLineTokenizerKeyVal oper = new FilteredLineTokenizerKeyVal();
  HashTestSink tokenSink = new HashTestSink();
  oper.setSplitBy(",");
  oper.setSplitTokenBy("=");
  oper.tokens.setSink(tokenSink);
  String[] filters = new String[2];
  filters[0] = "a";
  filters[1] = "c";
  oper.setFilterBy(filters);
  oper.beginWindow(0); //
  String input1 = "a=2,b=3,c=4";
  String input2 = "d=2";
  String input3 = "";
  int numTuples = 1000;
  for (int i = 0; i < numTuples; i++) {
    oper.data.process(input1);
    oper.data.process(input2);
    oper.data.process(input3);
  }
  oper.endWindow(); //
  Assert.assertEquals("number emitted tuples", 2, tokenSink.map.size());
  HashMap<Object, Object> smap = tokenSink.map;
  for (Map.Entry<Object, Object> e : smap.entrySet()) {
    HashMap<String, String> kmap = (HashMap<String, String>)e.getKey();
    for (Map.Entry<String, String> o : kmap.entrySet()) {
      String key = o.getKey();
      String val = o.getValue();
      Assert.assertTrue(!key.equals("b"));
      Assert.assertTrue(!key.equals("d"));
      if (key.equals("a")) {
        Assert.assertEquals("value of \"a\"", "2", val);
      } else if (key.equals("c")) {
        Assert.assertEquals("value of \"c\"", "4", val);
      }
    }
  }
}