@Override
    public int writeTo(TransferableChannel destChannel, int previouslyWritten, int remaining) throws IOException {
        long position = this.position + previouslyWritten;
        int count = Math.min(remaining, sizeInBytes() - previouslyWritten);
        // safe to cast to int since `count` is an int
        return (int) destChannel.transferFrom(channel, position, count);
    }
@Test
    public void testWriteTo() throws IOException {

        org.apache.kafka.common.requests.ByteBufferChannel channel = new org.apache.kafka.common.requests.ByteBufferChannel(fileRecords.sizeInBytes());
        int size = fileRecords.sizeInBytes();

        UnalignedFileRecords records1 = fileRecords.sliceUnaligned(0, size / 2);
        UnalignedFileRecords records2 = fileRecords.sliceUnaligned(size / 2, size - size / 2);

        records1.writeTo(channel, 0, records1.sizeInBytes());
        records2.writeTo(channel, 0, records2.sizeInBytes());

        channel.close();
        Iterator<Record> records = MemoryRecords.readableRecords(channel.buffer()).records().iterator();
        for (byte[] value : values) {
            assertTrue(records.hasNext());
            assertEquals(records.next().value(), ByteBuffer.wrap(value));
        }
    }