def METHOD_NAME(
    context, config: AirbyteSyncConfig, airbyte: BaseAirbyteResource
) -> Iterable[Any]:
    """Executes a Airbyte job sync for a given ``connection_id``, and polls until that sync
    completes, raising an error if it is unsuccessful. It outputs a AirbyteOutput which contains
    the job details for a given ``connection_id``.
    It requires the use of the :py:class:`~dagster_airbyte.airbyte_resource`, which allows it to
    communicate with the Airbyte API.
    Examples:
        .. code-block:: python
            from dagster import job
            from dagster_airbyte import airbyte_resource, airbyte_sync_op
            my_airbyte_resource = airbyte_resource.configured(
                {
                    "host": {"env": "AIRBYTE_HOST"},
                    "port": {"env": "AIRBYTE_PORT"},
                }
            )
            sync_foobar = airbyte_sync_op.configured({"connection_id": "foobar"}, name="sync_foobar")
            @job(resource_defs={"airbyte": my_airbyte_resource})
            def my_simple_airbyte_job():
                sync_foobar()
            @job(resource_defs={"airbyte": my_airbyte_resource})
            def my_composed_airbyte_job():
                final_foobar_state = sync_foobar(start_after=some_op())
                other_op(final_foobar_state)
    """
    airbyte_output = airbyte.sync_and_poll(
        connection_id=config.connection_id,
        poll_interval=config.poll_interval,
        poll_timeout=config.poll_timeout,
    )
    if config.yield_materializations:
        yield from generate_materializations(
            airbyte_output, asset_key_prefix=config.asset_key_prefix
        )
    yield Output(
        airbyte_output,
        metadata={
            **_get_attempt(airbyte_output.job_details.get("attempts", [{}])[-1]).get(
                "totalStats", {}
            )
        },
    )