@Override
    public void ensureValid() {
        if (sizeInBytes() < RECORD_BATCH_OVERHEAD)
            throw new CorruptRecordException("Record batch is corrupt (the size " + sizeInBytes() +
                    " is smaller than the minimum allowed overhead " + RECORD_BATCH_OVERHEAD + ")");

        if (!isValid())
            throw new CorruptRecordException("Record is corrupt (stored crc = " + checksum()
                    + ", computed crc = " + computeChecksum() + ")");
    }
RecordBatch implementation for magic 2 and above. The schema is given below:

RecordBatch =>
 BaseOffset => Int64
 Length => Int32
 PartitionLeaderEpoch => Int32
 Magic => Int8
 CRC => Uint32
 Attributes => Int16
 LastOffsetDelta => Int32 // also serves as LastSequenceDelta
 BaseTimestamp => Int64
 MaxTimestamp => Int64
 ProducerId => Int64
 ProducerEpoch => Int16
 BaseSequence => Int32
 Records => [Record]

Note that when compression is enabled (see attributes below), the compressed record data is serialized
directly following the count of the number of records.

The CRC covers the data from the attributes to the end of the batch (i.e. all the bytes that follow the CRC). It is
located after the magic byte, which means that clients must parse the magic byte before deciding how to interpret
the bytes between the batch length and the magic byte. The partition leader epoch field is not included in the CRC
computation to avoid the need to recompute the CRC when this field is assigned for every batch that is received by
the broker. The CRC-32C (Castagnoli) polynomial is used for the computation.

On Compaction: Unlike the older message formats, magic v2 and above preserves the first and last offset/sequence
numbers from the original batch when the log is cleaned. This is required in order to be able to restore the
producer's state when the log is reloaded. If we did not retain the last sequence number, then following
a partition leader failure, once the new leader has rebuilt the producer state from the log, the next sequence
expected number would no longer be in sync with what was written by the client. This would cause an
unexpected OutOfOrderSequence error, which is typically fatal. The base sequence number must be preserved for
duplicate checking: the broker checks incoming Produce requests for duplicates by verifying that the first and
last sequence numbers of the incoming batch match the last from that producer.

Note that if all of the records in a batch are removed during compaction, the broker may still retain an empty
batch header in order to preserve the producer sequence information as described above. These empty batches
are retained only until either a new sequence number is written by the corresponding producer or the producerId
is expired from lack of activity.

There is no similar need to preserve the timestamp from the original batch after compaction. The BaseTimestamp
field therefore reflects the timestamp of the first record in the batch in most cases. If the batch is empty, the
BaseTimestamp will be set to -1 (NO_TIMESTAMP). If the delete horizon flag is set to 1, the BaseTimestamp
will be set to the time at which tombstone records and aborted transaction markers in the batch should be removed.

Similarly, the MaxTimestamp field reflects the maximum timestamp of the current records if the timestamp type
is CREATE_TIME. For LOG_APPEND_TIME, on the other hand, the MaxTimestamp field reflects the timestamp set
by the broker and is preserved after compaction. Additionally, the MaxTimestamp of an empty batch always retains
the previous value prior to becoming empty.

The current attributes are given below:

 ---------------------------------------------------------------------------------------------------------------------------
 | Unused (7-15) | Delete Horizon Flag (6) | Control (5) | Transactional (4) | Timestamp Type (3) | Compression Type (0-2) |
 ---------------------------------------------------------------------------------------------------------------------------
clients/src/main/java/org/apache/kafka/common/record/DefaultRecordBatch.java