(self, process_records_input)
Called by a KCLProcess with a list of records to be processed and a checkpointer which accepts sequence numbers
        from the records to indicate where in the stream to checkpoint.

        :param amazon_kclpy.messages.ProcessRecordsInput process_records_input: the records, and metadata about the
            records.
Called by a KCLProcess with a list of records to be processed and a checkpointer which accepts sequence numbers
        from the records to indicate where in the stream to checkpoint.
Called by a KCLProcess with a list of records to be processed and a checkpointer which accepts sequence numbers from the records to indicate where in the stream to checkpoint .
def process_records(self, process_records_input):
        """
        Called by a KCLProcess with a list of records to be processed and a checkpointer which accepts sequence numbers
        from the records to indicate where in the stream to checkpoint.

        :param amazon_kclpy.messages.ProcessRecordsInput process_records_input: the records, and metadata about the
            records.
        """
        try:
            for record in process_records_input.records:
                data = record.binary_data
                seq = int(record.sequence_number)
                sub_seq = record.sub_sequence_number
                key = record.partition_key
                self.process_record(data, key, seq, sub_seq)
                if self.should_update_sequence(seq, sub_seq):
                    self._largest_seq = (seq, sub_seq)

            #
            # Checkpoints every self._CHECKPOINT_FREQ_SECONDS seconds
            #
            if time.time() - self._last_checkpoint_time > self._CHECKPOINT_FREQ_SECONDS:
                self.checkpoint(process_records_input.checkpointer, str(self._largest_seq[0]), self._largest_seq[1])
                self._last_checkpoint_time = time.time()

        except Exception as e:
            self.log("Encountered an exception while processing records. Exception was {e}\n".format(e=e))
def process_records ( self , process_records_input ) : try : for record in process_records_input . records : data = record . binary_data seq = int ( record . sequence_number ) sub_seq = record . sub_sequence_number key = record . partition_key self . process_record ( data , key , seq , sub_seq ) if self . should_update_sequence ( seq , sub_seq ) : self . _largest_seq = ( seq , sub_seq ) # # Checkpoints every self._CHECKPOINT_FREQ_SECONDS seconds # if time . time ( ) - self . _last_checkpoint_time > self . _CHECKPOINT_FREQ_SECONDS : self . checkpoint ( process_records_input . checkpointer , str ( self . _largest_seq [ 0 ] ) , self . _largest_seq [ 1 ] ) self . _last_checkpoint_time = time . time ( ) except Exception as e : self . log ( "Encountered an exception while processing records. Exception was {e}\n" . format ( e = e ) )