@Override
    public int writeTo(TransferableChannel channel, int previouslyWritten, int remaining) throws IOException {
        if (convertedRecordsWriter == null || convertedRecordsWriter.completed()) {
            MemoryRecords convertedRecords;

            try {
                // Check if we have more chunks left to down-convert
                if (convertedRecordsIterator.hasNext()) {
                    // Get next chunk of down-converted messages
                    ConvertedRecords<?> recordsAndStats = convertedRecordsIterator.next();
                    convertedRecords = (MemoryRecords) recordsAndStats.records();
                    recordValidationStats.add(recordsAndStats.recordConversionStats());
                    log.debug("Down-converted records for partition {} with length={}", topicPartition(), convertedRecords.sizeInBytes());
                } else {
                    convertedRecords = buildOverflowBatch(remaining);
                }
            } catch (UnsupportedCompressionTypeException e) {
                // We have encountered a compression type which does not support down-conversion (e.g. zstd).
                // Since we have already sent at least one batch and we have committed to the fetch size, we
                // send an overflow batch. The consumer will read the first few records and then fetch from the
                // offset of the batch which has the unsupported compression type. At that time, we will
                // send back the UNSUPPORTED_COMPRESSION_TYPE error which will allow the consumer to fail gracefully.
                convertedRecords = buildOverflowBatch(remaining);
            }

            convertedRecordsWriter = new DefaultRecordsSend<>(convertedRecords, Math.min(convertedRecords.sizeInBytes(), remaining));
        }
        // safe to cast to int since `remaining` is an int
        return (int) convertedRecordsWriter.writeTo(channel);
    }
Encapsulation for {@link RecordsSend} for {@link LazyDownConversionRecords}. Records are down-converted in batches and
on-demand when {@link #writeTo} method is called.
clients/src/main/java/org/apache/kafka/common/record/LazyDownConversionRecordsSend.java