@Test public void input_empty() throws Exception { LocalFileSystem fs = FileSystem.getLocal(conf); Path path = new Path(folder.newFile("testing").toURI()); fs.create(path).close(); try (ModelInput<StringOption> in = format.createInput( StringOption.class, fs, path, 0, fs.getFileStatus(path).getLen(), new Counter())) { assertThat("eof", in.readTo(new StringOption()), is(false)); } }
@Override public ModelInput<T> createInput( Class<? extends T> dataType, FileSystem fileSystem, Path path, long offset, long fragmentSize, Counter counter) throws IOException, InterruptedException { long end = offset + fragmentSize; K keyBuffer = createKeyObject(); V valueBuffer = createValueObject(); SequenceFile.Reader reader; try { reader = new SequenceFile.Reader(getConf(), SequenceFile.Reader.file(fileSystem.makeQualified(path))); } catch (EOFException e) { FileStatus status = fileSystem.getFileStatus(path); if (status.getLen() == 0L) { LOG.warn(MessageFormat.format( "Target sequence file is empty: {0}", path)); return new ModelInput<T>() { @Override public boolean readTo(T model) throws IOException { return false; } @Override public void close() throws IOException { return; } }; } throw e; } boolean succeed = false; try { if (offset > reader.getPosition()) { reader.sync(offset); } ModelInput<T> result = new ModelInput<T>() { private boolean next = reader.getPosition() < end; private long lastPosition = reader.getPosition(); @Override public boolean readTo(T model) throws IOException { if (next == false) { return false; } long current = reader.getPosition(); @SuppressWarnings("unchecked") K key = (K) reader.next(keyBuffer); if (key == null || (current >= end && reader.syncSeen())) { next = false; return false; } else { reader.getCurrentValue(valueBuffer); SequenceFileFormat.this.copyToModel(keyBuffer, valueBuffer, model); long nextPosition = reader.getPosition(); counter.add(nextPosition - lastPosition); lastPosition = nextPosition; return true; } } @Override public void close() throws IOException { reader.close(); } }; succeed = true; return result; } finally { if (succeed == false) { reader.close(); } } }
SequenceFileFormat extends HadoopFileFormat<T> { @Override public ModelInput<T> createInput( Class<? extends T> dataType, FileSystem fileSystem, Path path, long offset, long fragmentSize, Counter counter) throws IOException, InterruptedException { long end = offset + fragmentSize; K keyBuffer = createKeyObject(); V valueBuffer = createValueObject(); SequenceFile.Reader reader; try { reader = new SequenceFile.Reader(getConf(), SequenceFile.Reader.file(fileSystem.makeQualified(path))); } catch (EOFException e) { FileStatus status = fileSystem.getFileStatus(path); if (status.getLen() == 0L) { LOG.warn(MessageFormat.format( "Target sequence file is empty: {0}", path)); return new ModelInput<T>() { @Override public boolean readTo(T model) throws IOException { return false; } @Override public void close() throws IOException { return; } }; } throw e; } boolean succeed = false; try { if (offset > reader.getPosition()) { reader.sync(offset); } ModelInput<T> result = new ModelInput<T>() { private boolean next = reader.getPosition() < end; private long lastPosition = reader.getPosition(); @Override public boolean readTo(T model) throws IOException { if (next == false) { return false; } long current = reader.getPosition(); @SuppressWarnings("unchecked") K key = (K) reader.next(keyBuffer); if (key == null || (current >= end && reader.syncSeen())) { next = false; return false; } else { reader.getCurrentValue(valueBuffer); SequenceFileFormat.this.copyToModel(keyBuffer, valueBuffer, model); long nextPosition = reader.getPosition(); counter.add(nextPosition - lastPosition); lastPosition = nextPosition; return true; } } @Override public void close() throws IOException { reader.close(); } }; succeed = true; return result; } finally { if (succeed == false) { reader.close(); } } } }
SequenceFileFormat extends HadoopFileFormat<T> { @Override public ModelInput<T> createInput( Class<? extends T> dataType, FileSystem fileSystem, Path path, long offset, long fragmentSize, Counter counter) throws IOException, InterruptedException { long end = offset + fragmentSize; K keyBuffer = createKeyObject(); V valueBuffer = createValueObject(); SequenceFile.Reader reader; try { reader = new SequenceFile.Reader(getConf(), SequenceFile.Reader.file(fileSystem.makeQualified(path))); } catch (EOFException e) { FileStatus status = fileSystem.getFileStatus(path); if (status.getLen() == 0L) { LOG.warn(MessageFormat.format( "Target sequence file is empty: {0}", path)); return new ModelInput<T>() { @Override public boolean readTo(T model) throws IOException { return false; } @Override public void close() throws IOException { return; } }; } throw e; } boolean succeed = false; try { if (offset > reader.getPosition()) { reader.sync(offset); } ModelInput<T> result = new ModelInput<T>() { private boolean next = reader.getPosition() < end; private long lastPosition = reader.getPosition(); @Override public boolean readTo(T model) throws IOException { if (next == false) { return false; } long current = reader.getPosition(); @SuppressWarnings("unchecked") K key = (K) reader.next(keyBuffer); if (key == null || (current >= end && reader.syncSeen())) { next = false; return false; } else { reader.getCurrentValue(valueBuffer); SequenceFileFormat.this.copyToModel(keyBuffer, valueBuffer, model); long nextPosition = reader.getPosition(); counter.add(nextPosition - lastPosition); lastPosition = nextPosition; return true; } } @Override public void close() throws IOException { reader.close(); } }; succeed = true; return result; } finally { if (succeed == false) { reader.close(); } } }  }
SequenceFileFormat extends HadoopFileFormat<T> { @Override public ModelInput<T> createInput( Class<? extends T> dataType, FileSystem fileSystem, Path path, long offset, long fragmentSize, Counter counter) throws IOException, InterruptedException { long end = offset + fragmentSize; K keyBuffer = createKeyObject(); V valueBuffer = createValueObject(); SequenceFile.Reader reader; try { reader = new SequenceFile.Reader(getConf(), SequenceFile.Reader.file(fileSystem.makeQualified(path))); } catch (EOFException e) { FileStatus status = fileSystem.getFileStatus(path); if (status.getLen() == 0L) { LOG.warn(MessageFormat.format( "Target sequence file is empty: {0}", path)); return new ModelInput<T>() { @Override public boolean readTo(T model) throws IOException { return false; } @Override public void close() throws IOException { return; } }; } throw e; } boolean succeed = false; try { if (offset > reader.getPosition()) { reader.sync(offset); } ModelInput<T> result = new ModelInput<T>() { private boolean next = reader.getPosition() < end; private long lastPosition = reader.getPosition(); @Override public boolean readTo(T model) throws IOException { if (next == false) { return false; } long current = reader.getPosition(); @SuppressWarnings("unchecked") K key = (K) reader.next(keyBuffer); if (key == null || (current >= end && reader.syncSeen())) { next = false; return false; } else { reader.getCurrentValue(valueBuffer); SequenceFileFormat.this.copyToModel(keyBuffer, valueBuffer, model); long nextPosition = reader.getPosition(); counter.add(nextPosition - lastPosition); lastPosition = nextPosition; return true; } } @Override public void close() throws IOException { reader.close(); } }; succeed = true; return result; } finally { if (succeed == false) { reader.close(); } } }  @Override long getMinimumFragmentSize(); @Override ModelInput<T> createInput(
            Class<? extends T> dataType,
            FileSystem fileSystem,
            Path path,
            long offset,
            long fragmentSize,
            Counter counter); @Override ModelOutput<T> createOutput(
            Class<? extends T> dataType,
            FileSystem fileSystem,
            Path path,
            Counter counter); CompressionCodec getCompressionCodec(Path path); }
SequenceFileFormat extends HadoopFileFormat<T> { @Override public ModelInput<T> createInput( Class<? extends T> dataType, FileSystem fileSystem, Path path, long offset, long fragmentSize, Counter counter) throws IOException, InterruptedException { long end = offset + fragmentSize; K keyBuffer = createKeyObject(); V valueBuffer = createValueObject(); SequenceFile.Reader reader; try { reader = new SequenceFile.Reader(getConf(), SequenceFile.Reader.file(fileSystem.makeQualified(path))); } catch (EOFException e) { FileStatus status = fileSystem.getFileStatus(path); if (status.getLen() == 0L) { LOG.warn(MessageFormat.format( "Target sequence file is empty: {0}", path)); return new ModelInput<T>() { @Override public boolean readTo(T model) throws IOException { return false; } @Override public void close() throws IOException { return; } }; } throw e; } boolean succeed = false; try { if (offset > reader.getPosition()) { reader.sync(offset); } ModelInput<T> result = new ModelInput<T>() { private boolean next = reader.getPosition() < end; private long lastPosition = reader.getPosition(); @Override public boolean readTo(T model) throws IOException { if (next == false) { return false; } long current = reader.getPosition(); @SuppressWarnings("unchecked") K key = (K) reader.next(keyBuffer); if (key == null || (current >= end && reader.syncSeen())) { next = false; return false; } else { reader.getCurrentValue(valueBuffer); SequenceFileFormat.this.copyToModel(keyBuffer, valueBuffer, model); long nextPosition = reader.getPosition(); counter.add(nextPosition - lastPosition); lastPosition = nextPosition; return true; } } @Override public void close() throws IOException { reader.close(); } }; succeed = true; return result; } finally { if (succeed == false) { reader.close(); } } }  @Override long getMinimumFragmentSize(); @Override ModelInput<T> createInput(
            Class<? extends T> dataType,
            FileSystem fileSystem,
            Path path,
            long offset,
            long fragmentSize,
            Counter counter); @Override ModelOutput<T> createOutput(
            Class<? extends T> dataType,
            FileSystem fileSystem,
            Path path,
            Counter counter); CompressionCodec getCompressionCodec(Path path);  }