CommitMarkerCodec implements Closeable { public int readMarker(SequenceFile.Reader reader) throws IOException { if (valueBytes == null) { valueBytes = reader.createValueBytes(); } rawKey.reset(); rawValue.reset(); int status = reader.nextRaw(rawKey, valueBytes); if (status == -1) { return -1; } if (isMarkerValid()) { valueBytes.writeUncompressedBytes(rawValue); rawValue.flush(); return Ints.fromByteArray(rawValue.getData()); } throw new IOException(String.format("Invalid key for num entries appended found %s, expected : %s", new String(rawKey.getData()), TxConstants.TransactionLog.NUM_ENTRIES_APPENDED)); } CommitMarkerCodec(); @Override void close(); int readMarker(SequenceFile.Reader reader); static void writeMarker(SequenceFile.Writer writer, int count);  }
@Test public void testIncompleteCommitMarker() throws Exception { Path newLog = new Path(TMP_FOLDER.newFolder().getAbsolutePath(), LOG_FILE); try (SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, newLog, LongWritable.class, LongWritable.class, SequenceFile.CompressionType.NONE)) { String key = TxConstants.TransactionLog.NUM_ENTRIES_APPENDED; SequenceFile.ValueBytes valueBytes = new IncompleteValueBytes(); writer.appendRaw(key.getBytes(), 0, key.length(), valueBytes); writer.hflush(); writer.hsync(); } try (SequenceFile.Reader reader = new SequenceFile.Reader(fs, newLog, conf); CommitMarkerCodec markerCodec = new CommitMarkerCodec()) { try { markerCodec.readMarker(reader); Assert.fail("Expected EOF Exception to be thrown"); } catch (EOFException e) { } } }
@Test public void testIncorrectCommitMarker() throws Exception { Path newLog = new Path(TMP_FOLDER.newFolder().getAbsolutePath(), LOG_FILE); try (SequenceFile.Writer writer = SequenceFile.createWriter(fs, conf, newLog, LongWritable.class, LongWritable.class, SequenceFile.CompressionType.NONE)) { String invalidKey = "IncorrectKey"; SequenceFile.ValueBytes valueBytes = new CommitMarkerCodec.CommitEntriesCount(100); writer.appendRaw(invalidKey.getBytes(), 0, invalidKey.length(), valueBytes); writer.hflush(); writer.hsync(); } try (SequenceFile.Reader reader = new SequenceFile.Reader(fs, newLog, conf); CommitMarkerCodec markerCodec = new CommitMarkerCodec()) { try { markerCodec.readMarker(reader); Assert.fail("Expected an IOException to be thrown"); } catch (IOException e) { } } }