def _CallFlowRelational(self,
                          flow_name=None,
                          args=None,
                          runner_args=None,
                          client_id=None,
                          **kwargs):
    """Creates a new flow and send its responses to a state.

    This creates a new flow. The flow may send back many responses which will be
    queued by the framework until the flow terminates. The final status message
    will cause the entire transaction to be committed to the specified state.

    Args:
       flow_name: The name of the flow to invoke.
       args: Flow arguments.
       runner_args: Flow runner arguments.
       client_id: If given, the flow is started for this client.
       **kwargs: Arguments for the child flow.

    Returns:
       The URN of the child flow which was created.

    Raises:
       RuntimeError: In case of no cpu quota left to start more clients.
    """
    if isinstance(client_id, rdfvalue.RDFURN):
      client_id = client_id.Basename()

    if flow_name is None and runner_args is not None:
      flow_name = runner_args.flow_name

    flow_cls = registry.FlowRegistry.FlowClassByName(flow_name)

    flow_id = flow.StartFlow(
        client_id=client_id,
        creator=self.hunt_obj.creator,
        cpu_limit=self._GetSubFlowCPULimit(),
        network_bytes_limit=self._GetSubFlowNetworkLimit(),
        flow_cls=flow_cls,
        flow_args=args,
        parent_hunt_id=self.hunt_obj.urn.Basename(),
        **kwargs)

    return rdfvalue.RDFURN(client_id).Add("flows").Add(flow_id)