KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use should_update_sequence method in localstack

Best Python code snippet using localstack_python

sample_kclpy_app.py

Source:sample_kclpy_app.py

...79        # Insert your processing logic here80        ####################################81        self.log("Record (Partition Key: {pk}, Sequence Number: {seq}, Subsequence Number: {sseq}, Data Size: {ds}"82                 .format(pk=partition_key, seq=sequence_number, sseq=sub_sequence_number, ds=len(data)))83    def should_update_sequence(self, sequence_number, sub_sequence_number):84        """85        Determines whether a new larger sequence number is available86        :param int sequence_number: the sequence number from the current record87        :param int sub_sequence_number: the sub sequence number from the current record88        :return boolean: true if the largest sequence should be updated, false otherwise89        """90        return self._largest_seq == (None, None) or sequence_number > self._largest_seq[0] or \91            (sequence_number == self._largest_seq[0] and sub_sequence_number > self._largest_seq[1])92    def process_records(self, process_records_input):93        """94        Called by a KCLProcess with a list of records to be processed and a checkpointer which accepts sequence numbers95        from the records to indicate where in the stream to checkpoint.96        :param amazon_kclpy.messages.ProcessRecordsInput process_records_input: the records, and metadata about the97            records.98        """99        try:100            for record in process_records_input.records:101                data = record.binary_data102                seq = int(record.sequence_number)103                sub_seq = record.sub_sequence_number104                key = record.partition_key105                self.process_record(data, key, seq, sub_seq)106                if self.should_update_sequence(seq, sub_seq):107                    self._largest_seq = (seq, sub_seq)108            #109            # Checkpoints every self._CHECKPOINT_FREQ_SECONDS seconds110            #111            if time.time() - self._last_checkpoint_time > self._CHECKPOINT_FREQ_SECONDS:112                self.checkpoint(process_records_input.checkpointer, str(self._largest_seq[0]), self._largest_seq[1])113                self._last_checkpoint_time = time.time()114        except Exception as e:115            self.log("Encountered an exception while processing records. Exception was {e}\n".format(e=e))116    def lease_lost(self, lease_lost_input):117        self.log("Lease has been lost")118    def shard_ended(self, shard_ended_input):119        self.log("Shard has ended checkpointing")120        shard_ended_input.checkpointer.checkpoint()...

base.py

Source:base.py

...35                    stderr.write('MultiLangDaemon reported an invalid state while checkpointing.\n')36                else: # Some other error37                    stderr.write('ncountered an error while checkpointing, error was %s.\n' % e)38            sleep(self.checkpoint_retry_wait)39    def should_update_sequence(self, sequence_number, sub_sequence_number):40        return self.largest_seq == (None, None) or sequence_number > self.largest_seq[0] or \41            (sequence_number == self.largest_seq[0] and sub_sequence_number > self.largest_seq[1])42    def process_record(self, data, partition_key, sequence_number, sub_sequence_number):43        pass44    def process_records(self, process_records_input):45        try:46            for record in process_records_input.records:47                data = record.binary_data48                key = record.partition_key49                seq = int(record.sequence_number)50                sub_seq = record.sub_sequence_number51                self.process_record(data, key, seq, sub_seq)52                if self.should_update_sequence(seq, sub_seq):53                    self.largest_seq = (seq, sub_seq)54            if time() - self.last_checkpoint_time > self.checkpoint_seconds:55                self.checkpoint(process_records_input.checkpointer, str(self.largest_seq[0]), self.largest_seq[1])56                self.last_checkpoint_time = time()57        except Exception as e:58            # catch the exception, checkpoint, and refuse to continue59            stderr.write("Encountered an exception while processing records. Exception was %s\n" % e)60            self.checkpoint(process_records_input.checkpointer, str(self.largest_seq[0]), self.largest_seq[1])61            self.last_checkpoint_time = time()62            exit(1)63    def shutdown(self, shutdown_input):64        try:65            if shutdown_input.reason == 'TERMINATE':66                print('Was told to terminate, will attempt to checkpoint.')...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.