ExternalProcessInputFormat extends GenericInputFormat { @Override public void open(GenericInputSplit split) throws IOException { if(!(split instanceof ExternalProcessInputSplit)) { throw new IOException("Invalid InputSplit type."); } ExternalProcessInputSplit epSplit = (ExternalProcessInputSplit)split; if(epSplit.getExternalProcessCommand() != null && !epSplit.getExternalProcessCommand().equals("")) { try { this.extProc = Runtime.getRuntime().exec(epSplit.getExternalProcessCommand()); } catch (IOException e) { throw new IOException("IO Exception when starting external process: "+epSplit.getExternalProcessCommand()); } this.extProcOutStream = this.extProc.getInputStream(); this.extProcErrStream = this.extProc.getErrorStream(); } else { throw new IllegalArgumentException("External Process Command not set"); } }  @Override void configure(Configuration parameters); @Override void close(); @Override void open(GenericInputSplit split); void waitForProcessToFinish(); static final String ALLOWEDEXITCODES_PARAMETER_KEY; }
@Test public void testOpen() { Configuration config = new Configuration(); ExternalProcessInputSplit split = new ExternalProcessInputSplit(1, this.neverEndingCommand); boolean processDestroyed = false; try { format.configure(config); format.open(split); String[] cmd = {"/bin/sh","-c","ps aux | grep -v grep | grep \"cat /dev/urandom\" | wc -l"}; byte[] wcOut = new byte[128]; Process p = Runtime.getRuntime().exec(cmd); p.getInputStream().read(wcOut); int pCnt = Integer.parseInt(new String(wcOut).trim()); Assert.assertTrue(pCnt > 0); format.close(); } catch (IOException e) { Assert.fail(); } catch (RuntimeException e) { if(e.getMessage().equals("External process was destroyed although stream was not fully read.")) { processDestroyed = true; } } finally { Assert.assertTrue(processDestroyed); } }