@Test public void testDataSinkTask() { FileReader fr = null; BufferedReader br = null; try { int keyCnt = 100; int valCnt = 20; super.initEnvironment(MEMORY_MANAGER_SIZE, NETWORK_BUFFER_SIZE); super.addInput(new UniformRecordGenerator(keyCnt, valCnt, false), 0); DataSinkTask<Record> testTask = new DataSinkTask<>(this.mockEnv); File tempTestFile = new File(tempFolder.getRoot(), UUID.randomUUID().toString()); super.registerFileOutputTask(MockOutputFormat.class, tempTestFile.toURI().toString()); testTask.invoke(); Assert.assertTrue("Temp output file does not exist", tempTestFile.exists()); fr = new FileReader(tempTestFile); br = new BufferedReader(fr); HashMap<Integer, HashSet<Integer>> keyValueCountMap = new HashMap<>(keyCnt); while (br.ready()) { String line = br.readLine(); Integer key = Integer.parseInt(line.substring(0, line.indexOf("_"))); Integer val = Integer.parseInt(line.substring(line.indexOf("_") + 1, line.length())); if (!keyValueCountMap.containsKey(key)) { keyValueCountMap.put(key, new HashSet<Integer>()); } keyValueCountMap.get(key).add(val); } Assert.assertTrue("Invalid key count in out file. Expected: " + keyCnt + " Actual: " + keyValueCountMap.keySet().size(), keyValueCountMap.keySet().size() == keyCnt); for (Integer key : keyValueCountMap.keySet()) { Assert.assertTrue("Invalid value count for key: " + key + ". Expected: " + valCnt + " Actual: " + keyValueCountMap.get(key).size(), keyValueCountMap.get(key).size() == valCnt); } } catch (Exception e) { e.printStackTrace(); Assert.fail(e.getMessage()); } finally { if (br != null) { try { br.close(); } catch (Throwable t) {} } if (fr != null) { try { fr.close(); } catch (Throwable t) {} } } }
public DataSinkTask(Environment environment) { super(environment); }
DataSinkTask extends AbstractInvokable { public DataSinkTask(Environment environment) { super(environment); } }
DataSinkTask extends AbstractInvokable { public DataSinkTask(Environment environment) { super(environment); } DataSinkTask(Environment environment); }
DataSinkTask extends AbstractInvokable { public DataSinkTask(Environment environment) { super(environment); } DataSinkTask(Environment environment); @Override void invoke(); @Override void cancel(); DistributedRuntimeUDFContext createRuntimeContext(); }
DataSinkTask extends AbstractInvokable { public DataSinkTask(Environment environment) { super(environment); } DataSinkTask(Environment environment); @Override void invoke(); @Override void cancel(); DistributedRuntimeUDFContext createRuntimeContext();  }