FilteredLineTokenizerKeyVal extends LineTokenizerKeyVal { public void setFilterBy(String[] list) { if (list != null) { for (String s : list) { filterBy.put(s, null); } } }  void setFilterBy(String[] list); @Override boolean validSubTokenKey(String subtok);  }
@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); } } } }