ExternalProcessFixedLengthInputFormat extends ExternalProcessInputFormat<T> { @Override public void open(GenericInputSplit split) throws IOException { super.open(split); if(recordLength > this.targetReadBufferSize) { this.readBuffer = new byte[recordLength]; } else if (this.targetReadBufferSize % recordLength == 0) { this.readBuffer = new byte[this.targetReadBufferSize]; } else { this.readBuffer = new byte[(recordLength - (this.targetReadBufferSize % recordLength)) + this.targetReadBufferSize]; } this.readBufferReadPos = 0; this.readBufferFillPos = 0; this.readBufferRemainSpace = readBuffer.length; this.noMoreStreamInput = false; this.noMoreRecordBuffers = false; }  abstract boolean readBytes(PactRecord target, byte[] buffer, int startPos); @Override void configure(Configuration parameters); void setTargetReadBufferSize(int targetReadBufferSize); @Override void open(GenericInputSplit split); @Override boolean reachedEnd(); @Override boolean nextRecord(PactRecord record); static final String RECORDLENGTH_PARAMETER_KEY; }
@Test public void testOpen() { Configuration config = new Configuration(); config.setInteger(ExternalProcessFixedLengthInputFormat.RECORDLENGTH_PARAMETER_KEY, 8); 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); } }