<s> from corehq . apps . change_feed import topics <EOL> from corehq . apps . change_feed . consumer . feed import KafkaChangeFeed <EOL> from corehq . elastic import get_es_new <EOL> from corehq . pillows . mappings . sms_mapping import SMS_INDEX_INFO <EOL> from pillowtop . checkpoints . manager import PillowCheckpoint , PillowCheckpointEventHandler <EOL> from pillowtop . pillow . interface import ConstructedPillow <EOL> from pillowtop . processors . elastic import ElasticProcessor <EOL> SMS_PILLOW_CHECKPOINT_ID = '<STR_LIT>' <EOL> SMS_PILLOW_KAFKA_CONSUMER_GROUP_ID = '<STR_LIT>' <EOL> def get_sql_sms_pillow ( pillow_id ) : <EOL> checkpoint = PillowCheckpoint ( SMS_PILLOW_CHECKPOINT_ID ) <EOL> processor = ElasticProcessor ( <EOL> elasticsearch = get_es_new ( ) , <EOL> index_info = SMS_INDEX_INFO , <EOL> doc_prep_fn = lambda x : x <EOL> ) <EOL> return ConstructedPillow ( <EOL> name = pillow_id , <EOL> checkpoint = checkpoint , <EOL> change_feed = KafkaChangeFeed ( topics = [ topics . SMS ] , group_id = SMS_PILLOW_KAFKA_CONSUMER_GROUP_ID ) , <EOL> processor = processor , <EOL> change_processed_event_handler = PillowCheckpointEventHandler ( <EOL> checkpoint = checkpoint , checkpoint_frequency = <NUM_LIT:100> , <EOL> ) , <EOL> ) </s>
