public static ByteBuffer serializeAssignment(Assignment assignment) {
        Struct struct = new Struct(ASSIGNMENT_V0);
        struct.set(ERROR_KEY_NAME, assignment.error());
        struct.set(LEADER_KEY_NAME, assignment.leader());
        struct.set(LEADER_URL_KEY_NAME, assignment.leaderUrl());
        struct.set(CONFIG_OFFSET_KEY_NAME, assignment.offset());
        List<Struct> taskAssignments = new ArrayList<>();
        for (Map.Entry<String, Collection<Integer>> connectorEntry : assignment.asMap().entrySet()) {
            Struct taskAssignment = new Struct(CONNECTOR_ASSIGNMENT_V0);
            taskAssignment.set(CONNECTOR_KEY_NAME, connectorEntry.getKey());
            Collection<Integer> tasks = connectorEntry.getValue();
            taskAssignment.set(TASKS_KEY_NAME, tasks.toArray());
            taskAssignments.add(taskAssignment);
        }
        struct.set(ASSIGNMENT_KEY_NAME, taskAssignments.toArray());

        ByteBuffer buffer = ByteBuffer.allocate(CONNECT_PROTOCOL_HEADER_V0.sizeOf() + ASSIGNMENT_V0.sizeOf(struct));
        CONNECT_PROTOCOL_HEADER_V0.writeTo(buffer);
        ASSIGNMENT_V0.write(buffer, struct);
        buffer.flip();
        return buffer;
    }
The fields are serialized in sequence as follows:
Complete Assignment V0:
<pre>
  Version            => Int16
  Error              => Int16
  Leader             => [String]
  LeaderUrl          => [String]
  ConfigOffset       => Int64
  Assignment         => [Connector Assignment]
</pre>
connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/ConnectProtocol.java