DataSourceTask extends AbstractInvokable { @Override public void invoke() throws Exception { initInputFormat(); LOG.debug(getLogString("Start registering input and output")); try { initOutputs(getUserCodeClassLoader()); } catch (Exception ex) { throw new RuntimeException("The initialization of the DataSource's outputs caused an error: " + ex.getMessage(), ex); } LOG.debug(getLogString("Finished registering input and output")); LOG.debug(getLogString("Starting data source operator")); if(RichInputFormat.class.isAssignableFrom(this.format.getClass())){ ((RichInputFormat) this.format).setRuntimeContext(createRuntimeContext()); LOG.debug(getLogString("Rich Source detected. Initializing runtime context.")); } ExecutionConfig executionConfig; try { ExecutionConfig c = (ExecutionConfig) InstantiationUtil.readObjectFromConfig( getJobConfiguration(), ExecutionConfig.CONFIG_KEY, getUserCodeClassLoader()); if (c != null) { executionConfig = c; } else { LOG.warn("ExecutionConfig from job configuration is null. Creating empty config"); executionConfig = new ExecutionConfig(); } } catch (IOException | ClassNotFoundException e) { throw new RuntimeException("Could not load ExecutionConfig from Job Configuration: ", e); } boolean objectReuseEnabled = executionConfig.isObjectReuseEnabled(); LOG.debug("DataSourceTask object reuse: " + (objectReuseEnabled ? "ENABLED" : "DISABLED") + "."); final TypeSerializer<OT> serializer = this.serializerFactory.getSerializer(); try { BatchTask.openChainedTasks(this.chainedTasks, this); final Iterator<InputSplit> splitIterator = getInputSplits(); while (!this.taskCanceled && splitIterator.hasNext()) { final InputSplit split = splitIterator.next(); LOG.debug(getLogString("Opening input split " + split.toString())); final InputFormat<OT, InputSplit> format = this.format; format.open(split); LOG.debug(getLogString("Starting to read input from split " + split.toString())); try { final Collector<OT> output = this.output; if (objectReuseEnabled) { OT reuse1 = serializer.createInstance(); OT reuse2 = serializer.createInstance(); OT reuse3 = serializer.createInstance(); while (!this.taskCanceled && !format.reachedEnd()) { OT returned; if ((returned = format.nextRecord(reuse1)) != null) { output.collect(returned); reuse1 = reuse2; reuse2 = reuse3; reuse3 = returned; } } } else { while (!this.taskCanceled && !format.reachedEnd()) { OT returned; if ((returned = format.nextRecord(serializer.createInstance())) != null) { output.collect(returned); } } } if (LOG.isDebugEnabled() && !this.taskCanceled) { LOG.debug(getLogString("Closing input split " + split.toString())); } } finally { format.close(); } } this.output.close(); BatchTask.closeChainedTasks(this.chainedTasks, this); } catch (Exception ex) { try { this.format.close(); } catch (Throwable ignored) {} BatchTask.cancelChainedTasks(this.chainedTasks); ex = ExceptionInChainedStubException.exceptionUnwrap(ex); if (ex instanceof CancelTaskException) { throw ex; } else if (!this.taskCanceled) { BatchTask.logAndThrowException(ex, this); } } finally { BatchTask.clearWriters(eventualOutputs); } if (!this.taskCanceled) { LOG.debug(getLogString("Finished data source operator")); } else { LOG.debug(getLogString("Data source operator cancelled")); } }  @Override void invoke(); @Override void cancel(); DistributedRuntimeUDFContext createRuntimeContext();  }
@Test public void testDataSourceTask() { int keyCnt = 100; int valCnt = 20; this.outList = new ArrayList<Record>(); try { InputFilePreparator.prepareInputFile(new UniformRecordGenerator(keyCnt, valCnt, false), this.tempTestPath, true); } catch (IOException e1) { Assert.fail("Unable to set-up test input file"); } super.initEnvironment(MEMORY_MANAGER_SIZE, NETWORK_BUFFER_SIZE); super.addOutput(this.outList); DataSourceTask<Record> testTask = new DataSourceTask<>(); super.registerFileInputTask(testTask, MockInputFormat.class, new File(tempTestPath).toURI().toString(), "\n"); try { testTask.invoke(); } catch (Exception e) { System.err.println(e); Assert.fail("Invoke method caused exception."); } Assert.assertTrue("Invalid output size. Expected: "+(keyCnt*valCnt)+" Actual: "+this.outList.size(), this.outList.size() == keyCnt * valCnt); HashMap<Integer,HashSet<Integer>> keyValueCountMap = new HashMap<>(keyCnt); for (Record kvp : this.outList) { int key = kvp.getField(0, IntValue.class).getValue(); int val = kvp.getField(1, IntValue.class).getValue(); 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 mapKey : keyValueCountMap.keySet()) { Assert.assertTrue("Invalid value count for key: "+mapKey+". Expected: "+valCnt+" Actual: "+keyValueCountMap.get(mapKey).size(), keyValueCountMap.get(mapKey).size() == valCnt); } }
@Test public void testFailingDataSourceTask() { int keyCnt = 20; int valCnt = 10; this.outList = new NirvanaOutputList(); try { InputFilePreparator.prepareInputFile(new UniformRecordGenerator(keyCnt, valCnt, false), this.tempTestPath, false); } catch (IOException e1) { Assert.fail("Unable to set-up test input file"); } super.initEnvironment(MEMORY_MANAGER_SIZE, NETWORK_BUFFER_SIZE); super.addOutput(this.outList); DataSourceTask<Record> testTask = new DataSourceTask<>(); super.registerFileInputTask(testTask, MockFailingInputFormat.class, new File(tempTestPath).toURI().toString(), "\n"); boolean stubFailed = false; try { testTask.invoke(); } catch (Exception e) { stubFailed = true; } Assert.assertTrue("Function exception was not forwarded.", stubFailed); File tempTestFile = new File(this.tempTestPath); Assert.assertTrue("Temp output file does not exist",tempTestFile.exists()); }
@Test public void testCancelDataSourceTask() { int keyCnt = 20; int valCnt = 4; super.initEnvironment(MEMORY_MANAGER_SIZE, NETWORK_BUFFER_SIZE); super.addOutput(new NirvanaOutputList()); try { InputFilePreparator.prepareInputFile(new UniformRecordGenerator(keyCnt, valCnt, false), this.tempTestPath, false); } catch (IOException e1) { Assert.fail("Unable to set-up test input file"); } final DataSourceTask<Record> testTask = new DataSourceTask<>(); super.registerFileInputTask(testTask, MockDelayingInputFormat.class, new File(tempTestPath).toURI().toString(), "\n"); Thread taskRunner = new Thread() { @Override public void run() { try { testTask.invoke(); } catch (Exception ie) { ie.printStackTrace(); Assert.fail("Task threw exception although it was properly canceled"); } } }; taskRunner.start(); TaskCancelThread tct = new TaskCancelThread(1, taskRunner, testTask); tct.start(); try { tct.join(); taskRunner.join(); } catch(InterruptedException ie) { Assert.fail("Joining threads failed"); } File tempTestFile = new File(this.tempTestPath); Assert.assertTrue("Temp output file does not exist",tempTestFile.exists()); }