(pillow_id='user-pillow', num_processes=1, dedicated_migration_process=False, process_num=0,
        skip_ucr=False, processor_chunk_size=DEFAULT_PROCESSOR_CHUNK_SIZE, **kwargs)
return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=change_feed,
        processor=[user_processor] if skip_ucr else [ucr_processor, user_processor],
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=change_feed
        ),
        processor_chunk_size=processor_chunk_size,
        process_num=process_num,
        is_dedicated_migration_process=dedicated_migration_process and (process_num == 0)
    )
Processes users and sends them to ES and UCRs.

    Processors:
      - :py:func:`pillowtop.processors.elastic.BulkElasticProcessor`
      - :py:func:`corehq.apps.userreports.pillow.ConfigurableReportPillowProcessor`
Processes users and sends them to ES and UCRs.
Processes users and sends them to ES and UCRs .
def get_user_pillow(pillow_id='user-pillow', num_processes=1, dedicated_migration_process=False, process_num=0,
        skip_ucr=False, processor_chunk_size=DEFAULT_PROCESSOR_CHUNK_SIZE, **kwargs):
    """Processes users and sends them to ES and UCRs.

    Processors:
      - :py:func:`pillowtop.processors.elastic.BulkElasticProcessor`
      - :py:func:`corehq.apps.userreports.pillow.ConfigurableReportPillowProcessor`
    """
    # Pillow that sends users to ES and UCR
    assert pillow_id == 'user-pillow', 'Pillow ID is not allowed to change'
    checkpoint = get_checkpoint_for_elasticsearch_pillow(pillow_id, USER_INDEX_INFO, topics.USER_TOPICS)
    user_processor = get_user_es_processor()
    ucr_processor = get_ucr_processor(
        data_source_providers=[
            DynamicDataSourceProvider('CommCareUser'),
            StaticDataSourceProvider('CommCareUser')
        ],
        run_migrations=(process_num == 0),  # only first process runs migrations,
    )
    change_feed = KafkaChangeFeed(
        topics=topics.USER_TOPICS, client_id='users-to-es', num_processes=num_processes, process_num=process_num,
        dedicated_migration_process=dedicated_migration_process
    )
    return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=change_feed,
        processor=[user_processor] if skip_ucr else [ucr_processor, user_processor],
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=change_feed
        ),
        processor_chunk_size=processor_chunk_size,
        process_num=process_num,
        is_dedicated_migration_process=dedicated_migration_process and (process_num == 0)
    )
def get_user_pillow ( pillow_id = 'user-pillow' , num_processes = 1 , dedicated_migration_process = False , process_num = 0 , skip_ucr = False , processor_chunk_size = DEFAULT_PROCESSOR_CHUNK_SIZE , * * kwargs ) : # Pillow that sends users to ES and UCR assert pillow_id == 'user-pillow' , 'Pillow ID is not allowed to change' checkpoint = get_checkpoint_for_elasticsearch_pillow ( pillow_id , USER_INDEX_INFO , topics . USER_TOPICS ) user_processor = get_user_es_processor ( ) ucr_processor = get_ucr_processor ( data_source_providers = [ DynamicDataSourceProvider ( 'CommCareUser' ) , StaticDataSourceProvider ( 'CommCareUser' ) ] , run_migrations = ( process_num == 0 ) , # only first process runs migrations, ) change_feed = KafkaChangeFeed ( topics = topics . USER_TOPICS , client_id = 'users-to-es' , num_processes = num_processes , process_num = process_num , dedicated_migration_process = dedicated_migration_process ) return ConstructedPillow ( name = pillow_id , checkpoint = checkpoint , change_feed = change_feed , processor = [ user_processor ] if skip_ucr else [ ucr_processor , user_processor ] , change_processed_event_handler = KafkaCheckpointEventHandler ( checkpoint = checkpoint , checkpoint_frequency = 100 , change_feed = change_feed ) , processor_chunk_size = processor_chunk_size , process_num = process_num , is_dedicated_migration_process = dedicated_migration_process and ( process_num == 0 ) )