def execute_runner_async(runner, workunit_factory=None, workunit_name=None, workunit_labels=None,
                         workunit_log_config=None):
  """Executes the given java runner asynchronously.

  We can't use 'with' here because the workunit_generator's __exit__ function
  must be called after the process exits, in the return_code_handler.
  The wrapper around process.wait() needs to handle the same exceptions
  as the contextmanager does, so we have code duplication.

  We're basically faking the 'with' call to deal with asynchronous
  results.

  If `workunit_factory` is supplied, does so in the context of a workunit.

  :param runner: the java runner to run
  :param workunit_factory: an optional callable that can produce a workunit context
  :param string workunit_name: an optional name for the work unit; defaults to the main
  :param list workunit_labels: an optional sequence of labels for the work unit
  :param WorkUnit.LogConfig workunit_log_config: an optional tuple of task options affecting reporting

  Returns a ProcessHandler to the java process that is spawned.
  Raises `pants.java.Executor.Error` if there was a problem launching java itself.
  """

  if not isinstance(runner, Executor.Runner):
    raise ValueError('The runner argument must be a java Executor.Runner instance, '
                     'given {} of type {}'.format(runner, type(runner)))

  if workunit_factory is None:
    return SubprocessProcessHandler(runner.spawn())
  else:
    workunit_labels = [
                        WorkUnitLabel.TOOL,
                        WorkUnitLabel.NAILGUN if isinstance(runner.executor, NailgunExecutor) else WorkUnitLabel.JVM
                      ] + (workunit_labels or [])

    workunit_generator = workunit_factory(name=workunit_name, labels=workunit_labels,
                                cmd=runner.cmd, log_config=workunit_log_config)
    workunit = workunit_generator.__enter__()
    process = runner.spawn(stdout=workunit.output('stdout'), stderr=workunit.output('stderr'))

    class WorkUnitProcessHandler(ProcessHandler):
      def wait(_, timeout=None):
        try:
          ret = process.wait(timeout=timeout)
          workunit.set_outcome(WorkUnit.FAILURE if ret else WorkUnit.SUCCESS)
          workunit_generator.__exit__(None, None, None)
          return ret
        except BaseException:
          if not workunit_generator.__exit__(*sys.exc_info()):
            raise

      def kill(_):
        return process.kill()

      def terminate(_):
        return process.terminate()

      def poll(_):
        return process.poll()

    return WorkUnitProcessHandler()
Executes the given java runner asynchronously.

  We can't use 'with' here because the workunit_generator's __exit__ function
  must be called after the process exits, in the return_code_handler.
  The wrapper around process.wait() needs to handle the same exceptions
  as the contextmanager does, so we have code duplication.

  We're basically faking the 'with' call to deal with asynchronous
  results.

  If `workunit_factory` is supplied, does so in the context of a workunit.

  :param runner: the java runner to run
  :param workunit_factory: an optional callable that can produce a workunit context
  :param string workunit_name: an optional name for the work unit; defaults to the main
  :param list workunit_labels: an optional sequence of labels for the work unit
  :param WorkUnit.LogConfig workunit_log_config: an optional tuple of task options affecting reporting

  Returns a ProcessHandler to the java process that is spawned.
  Raises `pants.java.Executor.Error` if there was a problem launching java itself.