(self, changes_chunk)
Process given chunk in batch mode first on batch-processors
            and only latter on serial processors one by one, so that
            docs can be fetched only once while batch processing in bulk
            and cached via change.get_document for serial processors.
            For the relookup to be avoided at least one batch processor under
            should use change.set_document after docs are fetched in bulk

            If there is an exception in chunked processing, falls back
            to serial processing.
Process given chunk in batch mode first on batch-processors
            and only latter on serial processors one by one, so that
            docs can be fetched only once while batch processing in bulk
            and cached via change.get_document for serial processors.
            For the relookup to be avoided at least one batch processor under
            should use change.set_document after docs are fetched in bulk
Process given chunk in batch mode first on batch - processors and only latter on serial processors one by one so that docs can be fetched only once while batch processing in bulk and cached via change . get_document for serial processors . For the relookup to be avoided at least one batch processor under should use change . set_document after docs are fetched in bulk
def _batch_process_with_error_handling(self, changes_chunk):
        """
        Process given chunk in batch mode first on batch-processors
            and only latter on serial processors one by one, so that
            docs can be fetched only once while batch processing in bulk
            and cached via change.get_document for serial processors.
            For the relookup to be avoided at least one batch processor under
            should use change.set_document after docs are fetched in bulk

            If there is an exception in chunked processing, falls back
            to serial processing.
        """
        processing_time = 0

        def reprocess_serially(chunk, processor):
            for change in chunk:
                self.process_with_error_handling(change, processor)

        for processor in self.batch_processors:
            if not changes_chunk:
                return set(), 0

            changes_chunk = self._deduplicate_changes(changes_chunk)
            timer = TimingContext()
            with timer:
                try:
                    retry_changes, change_exceptions = processor.process_changes_chunk(changes_chunk)
                except Exception as ex:
                    notify_exception(
                        None,
                        "{pillow_name} Error in processing changes chunk: {ex}".format(
                            pillow_name=self.get_name(),
                            ex=ex
                        ),
                        details={
                            'change_ids': [c.id for c in changes_chunk]
                        })
                    self._record_batch_exception_in_datadog(processor)
                    # fall back to processing one by one
                    reprocess_serially(changes_chunk, processor)
                else:
                    # fall back to processing one by one for failed changes
                    for change, exception in change_exceptions:
                        handle_pillow_error(self, change, exception)
                    reprocess_serially(retry_changes, processor)
                processing_time += timer.duration
        # process on serial_processors
        for change in changes_chunk:
            processing_time += self.process_with_error_handling(change)
        self._record_datadog_metrics(changes_chunk, processing_time)
def _batch_process_with_error_handling ( self , changes_chunk ) : processing_time = 0 def reprocess_serially ( chunk , processor ) : for change in chunk : self . process_with_error_handling ( change , processor ) for processor in self . batch_processors : if not changes_chunk : return set ( ) , 0 changes_chunk = self . _deduplicate_changes ( changes_chunk ) timer = TimingContext ( ) with timer : try : retry_changes , change_exceptions = processor . process_changes_chunk ( changes_chunk ) except Exception as ex : notify_exception ( None , "{pillow_name} Error in processing changes chunk: {ex}" . format ( pillow_name = self . get_name ( ) , ex = ex ) , details = { 'change_ids' : [ c . id for c in changes_chunk ] } ) self . _record_batch_exception_in_datadog ( processor ) # fall back to processing one by one reprocess_serially ( changes_chunk , processor ) else : # fall back to processing one by one for failed changes for change , exception in change_exceptions : handle_pillow_error ( self , change , exception ) reprocess_serially ( retry_changes , processor ) processing_time += timer . duration # process on serial_processors for change in changes_chunk : processing_time += self . process_with_error_handling ( change ) self . _record_datadog_metrics ( changes_chunk , processing_time )