(pillow_id='LedgerToElasticsearchPillow', num_processes=1,
                                       process_num=0, **kwargs)
return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=change_feed,
        processor=LedgerProcessor(),
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=change_feed
        ),
    )
Ledger pillow

    Note that this pillow's id references Elasticsearch, but it no longer saves to ES.
    It has been kept to keep the checkpoint consistent, and can be changed at any time.

    Processors:
      - :py:class:`corehq.pillows.ledger.LedgerProcessor`
Ledger pillow
Ledger pillow
def get_ledger_to_elasticsearch_pillow(pillow_id='LedgerToElasticsearchPillow', num_processes=1,
                                       process_num=0, **kwargs):
    """Ledger pillow

    Note that this pillow's id references Elasticsearch, but it no longer saves to ES.
    It has been kept to keep the checkpoint consistent, and can be changed at any time.

    Processors:
      - :py:class:`corehq.pillows.ledger.LedgerProcessor`
    """
    assert pillow_id == 'LedgerToElasticsearchPillow', 'Pillow ID is not allowed to change'
    IndexInfo = namedtuple('IndexInfo', ['index'])
    checkpoint = get_checkpoint_for_elasticsearch_pillow(
        pillow_id, IndexInfo("ledgers_2016-03-15"), [topics.LEDGER]
    )
    change_feed = KafkaChangeFeed(
        topics=[topics.LEDGER], client_id='ledgers-to-es', num_processes=num_processes, process_num=process_num
    )
    return ConstructedPillow(
        name=pillow_id,
        checkpoint=checkpoint,
        change_feed=change_feed,
        processor=LedgerProcessor(),
        change_processed_event_handler=KafkaCheckpointEventHandler(
            checkpoint=checkpoint, checkpoint_frequency=100, change_feed=change_feed
        ),
    )
def get_ledger_to_elasticsearch_pillow ( pillow_id = 'LedgerToElasticsearchPillow' , num_processes = 1 , process_num = 0 , * * kwargs ) : assert pillow_id == 'LedgerToElasticsearchPillow' , 'Pillow ID is not allowed to change' IndexInfo = namedtuple ( 'IndexInfo' , [ 'index' ] ) checkpoint = get_checkpoint_for_elasticsearch_pillow ( pillow_id , IndexInfo ( "ledgers_2016-03-15" ) , [ topics . LEDGER ] ) change_feed = KafkaChangeFeed ( topics = [ topics . LEDGER ] , client_id = 'ledgers-to-es' , num_processes = num_processes , process_num = process_num ) return ConstructedPillow ( name = pillow_id , checkpoint = checkpoint , change_feed = change_feed , processor = LedgerProcessor ( ) , change_processed_event_handler = KafkaCheckpointEventHandler ( checkpoint = checkpoint , checkpoint_frequency = 100 , change_feed = change_feed ) , )