@Test(expected = IllegalArgumentException.class) public void truncateTest() throws IOException, InterruptedException { try (JournalPersistence store = prepareStore()) { int size = 10; int maxLength = 999; long start = store.max(); List<byte[]> journals = ByteUtils.createRandomSizeByteList(maxLength, size); int length = journals.stream().mapToInt(journal -> journal.length).sum(); long writePosition = 0L; for (byte[] journal : journals) { writePosition = store.append(journal); } Assert.assertEquals(length + start, writePosition); Assert.assertEquals(writePosition, store.max()); long position = IntStream.range(0, 7).mapToObj(journals::get).mapToInt(journal -> journal.length).sum(); store.truncate(position); Assert.assertEquals(position, store.max()); journals.remove(9); journals.remove(8); journals.remove(7); byte[] readBytes = store.read(start, length); byte[] writeBytes = ByteUtils.concatBytes(journals); Assert.assertArrayEquals(writeBytes, readBytes); while (store.flushed() < store.max()) { store.flush(); } journals.remove(6); position = IntStream.range(0, 6).mapToObj(journals::get).mapToInt(journal -> journal.length).sum(); store.truncate(position); readBytes = store.read(start, length); writeBytes = ByteUtils.concatBytes(journals); Assert.assertArrayEquals(writeBytes, readBytes); position += 1; store.truncate(position); } }
public void truncate(long givenMax) throws IOException { synchronized (fileMapMutex) { if (givenMax == max()) return; logger.info("Truncate to position: {}, min: {}, max: {}, flushed: {}, path: {}...", ThreadSafeFormat.formatWithComma(givenMax), ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()), ThreadSafeFormat.formatWithComma(flushPosition.get()), base.getAbsolutePath()); if (givenMax < min() || givenMax > max()) { throw new IllegalArgumentException( String.format("GivenMax %s should between [%s, %s]!", ThreadSafeFormat.formatWithComma(givenMax), ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()) ) ); } else if (givenMax < max()) { rollbackFiles(givenMax); this.writePosition.set(givenMax); if (this.flushPosition.get() > givenMax) this.flushPosition.set(givenMax); resetWriteStoreFile(); } } }
PositioningStore implements JournalPersistence, MonitoredPersistence, Closeable { public void truncate(long givenMax) throws IOException { synchronized (fileMapMutex) { if (givenMax == max()) return; logger.info("Truncate to position: {}, min: {}, max: {}, flushed: {}, path: {}...", ThreadSafeFormat.formatWithComma(givenMax), ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()), ThreadSafeFormat.formatWithComma(flushPosition.get()), base.getAbsolutePath()); if (givenMax < min() || givenMax > max()) { throw new IllegalArgumentException( String.format("GivenMax %s should between [%s, %s]!", ThreadSafeFormat.formatWithComma(givenMax), ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()) ) ); } else if (givenMax < max()) { rollbackFiles(givenMax); this.writePosition.set(givenMax); if (this.flushPosition.get() > givenMax) this.flushPosition.set(givenMax); resetWriteStoreFile(); } } } }
PositioningStore implements JournalPersistence, MonitoredPersistence, Closeable { public void truncate(long givenMax) throws IOException { synchronized (fileMapMutex) { if (givenMax == max()) return; logger.info("Truncate to position: {}, min: {}, max: {}, flushed: {}, path: {}...", ThreadSafeFormat.formatWithComma(givenMax), ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()), ThreadSafeFormat.formatWithComma(flushPosition.get()), base.getAbsolutePath()); if (givenMax < min() || givenMax > max()) { throw new IllegalArgumentException( String.format("GivenMax %s should between [%s, %s]!", ThreadSafeFormat.formatWithComma(givenMax), ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()) ) ); } else if (givenMax < max()) { rollbackFiles(givenMax); this.writePosition.set(givenMax); if (this.flushPosition.get() > givenMax) this.flushPosition.set(givenMax); resetWriteStoreFile(); } } } PositioningStore(); }
PositioningStore implements JournalPersistence, MonitoredPersistence, Closeable { public void truncate(long givenMax) throws IOException { synchronized (fileMapMutex) { if (givenMax == max()) return; logger.info("Truncate to position: {}, min: {}, max: {}, flushed: {}, path: {}...", ThreadSafeFormat.formatWithComma(givenMax), ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()), ThreadSafeFormat.formatWithComma(flushPosition.get()), base.getAbsolutePath()); if (givenMax < min() || givenMax > max()) { throw new IllegalArgumentException( String.format("GivenMax %s should between [%s, %s]!", ThreadSafeFormat.formatWithComma(givenMax), ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()) ) ); } else if (givenMax < max()) { rollbackFiles(givenMax); this.writePosition.set(givenMax); if (this.flushPosition.get() > givenMax) this.flushPosition.set(givenMax); resetWriteStoreFile(); } } } PositioningStore(); void truncate(long givenMax); void delete(); void recover(Path path, long min, Properties properties); @Override long append(byte[] bytes); @Override long append(List<byte[]> bytesList ); @Override long min(); @Override long physicalMin(); @Override long max(); @Override long flushed(); @Override void flush(); byte[] read(long position, int length); Long readLong(long position); long compact(long givenMin); @Override Path getBasePath(); @Override void close(); @Override Path getPath(); @Override long getFreeSpace(); @Override long getTotalSpace(); @Override String toString(); }
PositioningStore implements JournalPersistence, MonitoredPersistence, Closeable { public void truncate(long givenMax) throws IOException { synchronized (fileMapMutex) { if (givenMax == max()) return; logger.info("Truncate to position: {}, min: {}, max: {}, flushed: {}, path: {}...", ThreadSafeFormat.formatWithComma(givenMax), ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()), ThreadSafeFormat.formatWithComma(flushPosition.get()), base.getAbsolutePath()); if (givenMax < min() || givenMax > max()) { throw new IllegalArgumentException( String.format("GivenMax %s should between [%s, %s]!", ThreadSafeFormat.formatWithComma(givenMax), ThreadSafeFormat.formatWithComma(min()), ThreadSafeFormat.formatWithComma(max()) ) ); } else if (givenMax < max()) { rollbackFiles(givenMax); this.writePosition.set(givenMax); if (this.flushPosition.get() > givenMax) this.flushPosition.set(givenMax); resetWriteStoreFile(); } } } PositioningStore(); void truncate(long givenMax); void delete(); void recover(Path path, long min, Properties properties); @Override long append(byte[] bytes); @Override long append(List<byte[]> bytesList ); @Override long min(); @Override long physicalMin(); @Override long max(); @Override long flushed(); @Override void flush(); byte[] read(long position, int length); Long readLong(long position); long compact(long givenMin); @Override Path getBasePath(); @Override void close(); @Override Path getPath(); @Override long getFreeSpace(); @Override long getTotalSpace(); @Override String toString();  }