#NEWLINE# def get_sql_case_to_elasticsearch_pillow( pillow_id = 'SqlCaseToElasticsearchPillow' ) : #NEWLINE# #INDENT# checkpoint = PillowCheckpoint ( 'sql-cases-to-elasticsearch' ) #NEWLINE# case_processor = ElasticProcessor ( elasticsearch = get_es_new ( ) , index_info = ElasticsearchIndexInfo ( index = CASE_INDEX , type = CASE_ES_TYPE ) , doc_prep_fn = transform_case_for_elasticsearch ) #NEWLINE# return ConstructedPillow ( name = checkpoint , checkpoint = checkpoint , change_feed = KafkaChangeFeed ( topics = [ topics . CASE_SQL ] , group_id = 'sql-cases-to-es' ) , processor = case_processor , change_processed_event_handler = PillowCheckpointEventHandler ( checkpoint = checkpoint , checkpoint_frequency = 100 ) )