@Override
    public ByteBuffer toBuffer(int headerPos) {
        byte[] brokerRackBytes = brokerRack().getBytes(StandardCharsets.UTF_8);
        byte[] topic = topic().getBytes(StandardCharsets.UTF_8);
        ByteBuffer buffer = ByteBuffer.allocate(headerPos + Byte.BYTES
                + Long.BYTES
                + Integer.BYTES
                + Integer.BYTES
                + brokerRackBytes.length
                + Integer.BYTES
                + topic.length
                + Integer.BYTES
                + bodySize());
        buffer.position(headerPos);
        buffer.put(METRIC_VERSION);
        buffer.putLong(time());
        buffer.putInt(brokerId());
        buffer.putInt(brokerRackBytes.length);
        if (brokerRackBytes.length > 0) {
            buffer.put(brokerRackBytes);
        }
        buffer.putInt(topic.length);
        buffer.put(topic);
        buffer.putInt(partition);

        buffer = writeBody(buffer);
        return buffer;
    }
The buffer capacity is calculated as follows:
<ul>
  <li>(headerPos + {@link Byte#BYTES}) - version</li>
  <li>{@link Long#BYTES} - time</li>
  <li>{@link Integer#BYTES} - broker id</li>
  <li>{@link Integer#BYTES} - broker rack length</li>
  <li>brokerRack.length - broker rack</li>
  <li>{@link Integer#BYTES} - topic length</li>
  <li>topic.length - topic</li>
  <li>{@link Integer#BYTES} - partition</li>
  <li>body length - metric-value body</li>
</ul>

@param headerPos Header position
@return Byte buffer of the partition metric.
core/src/main/java/kafka/autobalancer/metricsreporter/metric/TopicPartitionMetrics.java