static EndTransactionMarker deserializeValue(ControlRecordType type, ByteBuffer value) {
        ensureTransactionMarkerControlType(type);

        if (value.remaining() < CURRENT_END_TXN_MARKER_VALUE_SIZE)
            throw new InvalidRecordException("Invalid value size found for end transaction marker. Must have " +
                    "at least " + CURRENT_END_TXN_MARKER_VALUE_SIZE + " bytes, but found only " + value.remaining());

        short version = value.getShort(0);
        if (version < 0)
            throw new InvalidRecordException("Invalid version found for end transaction marker: " + version +
                    ". May indicate data corruption");

        if (version > CURRENT_END_TXN_MARKER_VERSION)
            log.debug("Received end transaction marker value version {}. Parsing as version {}", version,
                    CURRENT_END_TXN_MARKER_VERSION);

        int coordinatorEpoch = value.getInt(2);
        return new EndTransactionMarker(type, coordinatorEpoch);
    }
@Test
    public void testNotEnoughBytes() {
        assertThrows(InvalidRecordException.class,
            () -> EndTransactionMarker.deserializeValue(ControlRecordType.COMMIT, ByteBuffer.wrap(new byte[0])));
    }