(pillow_id='XFormToElasticsearchPillow', num_processes=1,
                                      process_num=0, **kwargs)
return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=kafka_change_feed,
        processor=form_processor,
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=kafka_change_feed
        ),
    )
XForm change processor that sends form data to Elasticsearch

    Processors:
      - :py:class:`pillowtop.processors.elastic.ElasticProcessor`
XForm change processor that sends form data to Elasticsearch
XForm change processor that sends form data to Elasticsearch
def get_xform_to_elasticsearch_pillow(pillow_id='XFormToElasticsearchPillow', num_processes=1,
                                      process_num=0, **kwargs):
    """XForm change processor that sends form data to Elasticsearch

    Processors:
      - :py:class:`pillowtop.processors.elastic.ElasticProcessor`
    """
    # todo; To remove after full rollout of https://github.com/dimagi/commcare-hq/pull/21329/
    assert pillow_id == 'XFormToElasticsearchPillow', 'Pillow ID is not allowed to change'
    checkpoint = get_checkpoint_for_elasticsearch_pillow(pillow_id, XFORM_INDEX_INFO, FORM_TOPICS)
    form_processor = ElasticProcessor(
        elasticsearch=get_es_new(),
        index_info=XFORM_INDEX_INFO,
        doc_prep_fn=transform_xform_for_elasticsearch,
        doc_filter_fn=xform_pillow_filter,
        change_filter_fn=is_couch_change_for_sql_domain
    )
    kafka_change_feed = KafkaChangeFeed(
        topics=FORM_TOPICS, client_id='forms-to-es', num_processes=num_processes, process_num=process_num
    )
    return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=kafka_change_feed,
        processor=form_processor,
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=kafka_change_feed
        ),
    )
def get_xform_to_elasticsearch_pillow ( pillow_id = 'XFormToElasticsearchPillow' , num_processes = 1 , process_num = 0 , * * kwargs ) : # todo; To remove after full rollout of https://github.com/dimagi/commcare-hq/pull/21329/ assert pillow_id == 'XFormToElasticsearchPillow' , 'Pillow ID is not allowed to change' checkpoint = get_checkpoint_for_elasticsearch_pillow ( pillow_id , XFORM_INDEX_INFO , FORM_TOPICS ) form_processor = ElasticProcessor ( elasticsearch = get_es_new ( ) , index_info = XFORM_INDEX_INFO , doc_prep_fn = transform_xform_for_elasticsearch , doc_filter_fn = xform_pillow_filter , change_filter_fn = is_couch_change_for_sql_domain ) kafka_change_feed = KafkaChangeFeed ( topics = FORM_TOPICS , client_id = 'forms-to-es' , num_processes = num_processes , process_num = process_num ) return ConstructedPillow ( name = pillow_id , checkpoint = checkpoint , change_feed = kafka_change_feed , processor = form_processor , change_processed_event_handler = KafkaCheckpointEventHandler ( checkpoint = checkpoint , checkpoint_frequency = 100 , change_feed = kafka_change_feed ) , )