@Test public void testTruncateIfSizeIsDifferentToTargetSize() throws IOException { FileChannel channelMock = EasyMock.createMock(FileChannel.class); EasyMock.expect(channelMock.size()).andReturn(42L).atLeastOnce(); EasyMock.expect(channelMock.position(42L)).andReturn(null).once(); EasyMock.expect(channelMock.truncate(23L)).andReturn(null).once(); EasyMock.replay(channelMock); FileRecords fileRecords = new FileRecords(tempFile(), channelMock, 0, Integer.MAX_VALUE, false); fileRecords.truncateTo(23); EasyMock.verify(channelMock); }
public int truncateTo(int targetSize) throws IOException { int originalSize = sizeInBytes(); if (targetSize > originalSize || targetSize < 0) throw new KafkaException("Attempt to truncate log segment to " + targetSize + " bytes failed, " + " size of this log segment is " + originalSize + " bytes."); if (targetSize < (int) channel.size()) { channel.truncate(targetSize); size.set(targetSize); } return originalSize - targetSize; }
FileRecords extends AbstractRecords implements Closeable { public int truncateTo(int targetSize) throws IOException { int originalSize = sizeInBytes(); if (targetSize > originalSize || targetSize < 0) throw new KafkaException("Attempt to truncate log segment to " + targetSize + " bytes failed, " + " size of this log segment is " + originalSize + " bytes."); if (targetSize < (int) channel.size()) { channel.truncate(targetSize); size.set(targetSize); } return originalSize - targetSize; } }
FileRecords extends AbstractRecords implements Closeable { public int truncateTo(int targetSize) throws IOException { int originalSize = sizeInBytes(); if (targetSize > originalSize || targetSize < 0) throw new KafkaException("Attempt to truncate log segment to " + targetSize + " bytes failed, " + " size of this log segment is " + originalSize + " bytes."); if (targetSize < (int) channel.size()) { channel.truncate(targetSize); size.set(targetSize); } return originalSize - targetSize; } FileRecords(File file,
                       FileChannel channel,
                       int start,
                       int end,
                       boolean isSlice); }
FileRecords extends AbstractRecords implements Closeable { public int truncateTo(int targetSize) throws IOException { int originalSize = sizeInBytes(); if (targetSize > originalSize || targetSize < 0) throw new KafkaException("Attempt to truncate log segment to " + targetSize + " bytes failed, " + " size of this log segment is " + originalSize + " bytes."); if (targetSize < (int) channel.size()) { channel.truncate(targetSize); size.set(targetSize); } return originalSize - targetSize; } FileRecords(File file,
                       FileChannel channel,
                       int start,
                       int end,
                       boolean isSlice); @Override int sizeInBytes(); File file(); FileChannel channel(); ByteBuffer readInto(ByteBuffer buffer, int position); FileRecords read(int position, int size); int append(MemoryRecords records); void flush(); void close(); boolean delete(); void trim(); void setFile(File file); void renameTo(File f); int truncateTo(int targetSize); @Override Records downConvert(byte toMagic, long firstOffset); @Override long writeTo(GatheringByteChannel destChannel, long offset, int length); LogOffsetPosition searchForOffsetWithSize(long targetOffset, int startingPosition); TimestampAndOffset searchForTimestamp(long targetTimestamp, int startingPosition, long startingOffset); TimestampAndOffset largestTimestampAfter(int startingPosition); @Override Iterable<FileChannelRecordBatch> batches(); static FileRecords open(File file,
                                   boolean mutable,
                                   boolean fileAlreadyExists,
                                   int initFileSize,
                                   boolean preallocate); static FileRecords open(File file,
                                   boolean fileAlreadyExists,
                                   int initFileSize,
                                   boolean preallocate); static FileRecords open(File file, boolean mutable); static FileRecords open(File file); }
FileRecords extends AbstractRecords implements Closeable { public int truncateTo(int targetSize) throws IOException { int originalSize = sizeInBytes(); if (targetSize > originalSize || targetSize < 0) throw new KafkaException("Attempt to truncate log segment to " + targetSize + " bytes failed, " + " size of this log segment is " + originalSize + " bytes."); if (targetSize < (int) channel.size()) { channel.truncate(targetSize); size.set(targetSize); } return originalSize - targetSize; } FileRecords(File file,
                       FileChannel channel,
                       int start,
                       int end,
                       boolean isSlice); @Override int sizeInBytes(); File file(); FileChannel channel(); ByteBuffer readInto(ByteBuffer buffer, int position); FileRecords read(int position, int size); int append(MemoryRecords records); void flush(); void close(); boolean delete(); void trim(); void setFile(File file); void renameTo(File f); int truncateTo(int targetSize); @Override Records downConvert(byte toMagic, long firstOffset); @Override long writeTo(GatheringByteChannel destChannel, long offset, int length); LogOffsetPosition searchForOffsetWithSize(long targetOffset, int startingPosition); TimestampAndOffset searchForTimestamp(long targetTimestamp, int startingPosition, long startingOffset); TimestampAndOffset largestTimestampAfter(int startingPosition); @Override Iterable<FileChannelRecordBatch> batches(); static FileRecords open(File file,
                                   boolean mutable,
                                   boolean fileAlreadyExists,
                                   int initFileSize,
                                   boolean preallocate); static FileRecords open(File file,
                                   boolean fileAlreadyExists,
                                   int initFileSize,
                                   boolean preallocate); static FileRecords open(File file, boolean mutable); static FileRecords open(File file);  }