docstr: Encode msg data into the `msg_buffer`, which should be allocated
  to at least the size of this message.
code: def _encode_msg(
      self, start_pos, offset, timestamp, key, value, attributes=0
  ):
      """Encode msg data into the `msg_buffer`, which should be allocated
      to at least the size of this message.
      """
      magic = self._magic
      buf = self._buffer
      pos = start_pos
  
      # Write key and value
      pos += self.KEY_OFFSET_V0 if magic == 0 else self.KEY_OFFSET_V1
  
      if key is None:
          struct.pack_into(">i", buf, pos, -1)
          pos += self.KEY_LENGTH
      else:
          key_size = len(key)
          struct.pack_into(">i", buf, pos, key_size)
          pos += self.KEY_LENGTH
          buf[pos : pos + key_size] = key
          pos += key_size
  
      if value is None:
          struct.pack_into(">i", buf, pos, -1)
          pos += self.VALUE_LENGTH
      else:
          value_size = len(value)
          struct.pack_into(">i", buf, pos, value_size)
          pos += self.VALUE_LENGTH
          buf[pos : pos + value_size] = value
          pos += value_size
      length = (pos - start_pos) - self.LOG_OVERHEAD
  
      # Write msg header. Note, that Crc will be updated later
      if magic == 0:
          self.HEADER_STRUCT_V0.pack_into(
              buf, start_pos, offset, length, 0, magic, attributes
          )
      else:
          self.HEADER_STRUCT_V1.pack_into(
              buf,
              start_pos,
              offset,
              length,
              0,
              magic,
              attributes,
              timestamp,
          )
  
      # Calculate CRC for msg
      crc_data = memoryview(buf)[start_pos + self.MAGIC_OFFSET :]
      crc = calc_crc32(crc_data)
      struct.pack_into(">I", buf, start_pos + self.CRC_OFFSET, crc)
      return crc
  
imports: import struct
  import time
  from kafka.record.abc import ABCRecord, ABCRecordBatch, ABCRecordBatchBuilder
  from kafka.record.util import calc_crc32
  from kafka.codec import gzip_encode, snappy_encode, lz4_encode, lz4_encode_old_kafka, gzip_decode, snappy_decode, lz4_decode, lz4_decode_old_kafka
  import kafka.codec
  from kafka.errors import CorruptRecordException, UnsupportedCodecError
label_desc: Use image node_id 3 for calling the LegacyRecordBatchBuilder obj's underlying member method code with example usage: obj._encode_msg(start_pos, offset, timestamp, key, value, attributes) and returns: crc