def METHOD_NAME(task_runner):
    reset_processing_state()
    org = Factories.create_organization()
    with outbox_context(flush=False):
        ap = AuthProvider.objects.create(organization_id=org.id, provider="meethub", config={})
        for i in range(5):
            user = Factories.create_user()
            AuthIdentity.objects.create(user=user, auth_provider=ap, ident=str(i), data={})
    # Clear existing outboxes, force replication by hand
    ControlOutbox.objects.all().delete()
    assert not ControlOutbox.objects.all().exists()
    with assume_test_silo_mode(SiloMode.REGION):
        assert not AuthProviderReplica.objects.filter(auth_provider_id=ap.id).exists()
        assert not AuthIdentityReplica.objects.filter(auth_provider_id=ap.id).exists()
    with task_runner():
        schedule_backfill_outbox_jobs_control()
    assert (
        get_processing_state(AuthIdentity._meta.db_table)[1] == AuthIdentity.replication_version + 1
    )
    with outbox_runner():
        assert ControlOutbox.objects.all().count() == 6
        with assume_test_silo_mode(SiloMode.REGION):
            assert not AuthProviderReplica.objects.filter(auth_provider_id=ap.id).exists()
            assert not AuthIdentityReplica.objects.filter(auth_provider_id=ap.id).exists()
    with assume_test_silo_mode(SiloMode.REGION):
        assert AuthProviderReplica.objects.filter(auth_provider_id=ap.id).exists()
        assert AuthIdentityReplica.objects.filter(auth_provider_id=ap.id).count() == 5
    with outbox_context(flush=False):
        org2 = Factories.create_organization()
        ap2 = AuthProvider.objects.create(organization_id=org2.id, provider="meethub", config={})
        for i in range(5):
            user = Factories.create_user()
            AuthIdentity.objects.create(user=user, auth_provider=ap2, ident=str(i), data={})
    # Clear again
    ControlOutbox.objects.all().delete()
    with assume_test_silo_mode(SiloMode.REGION):
        assert AuthIdentityReplica.objects.filter(auth_provider_id=ap2.id).count() == 0
    with outbox_runner(), task_runner():
        schedule_backfill_outbox_jobs_control()
    # Does not process these new objects since we already completed all available work for this version.
    with assume_test_silo_mode(SiloMode.REGION):
        assert AuthIdentityReplica.objects.filter(auth_provider_id=ap2.id).count() == 0
        AuthIdentityReplica.objects.all().delete()
    with patch("sentry.models.authidentity.AuthIdentity.replication_version", new=10000):
        with outbox_runner(), task_runner():
            schedule_backfill_outbox_jobs_control()
        # Replicates it now that the version has bumped
        with assume_test_silo_mode(SiloMode.REGION):
            assert AuthIdentityReplica.objects.all().count() == 10
            assert AuthIdentityReplica.objects.filter(auth_provider_id=ap2.id).count() == 5
            assert AuthIdentityReplica.objects.filter(auth_provider_id=ap.id).count() == 5
        assert get_processing_state(AuthIdentity._meta.db_table)[1] == 10001