(self, checkpointer, sequence_number=None, sub_sequence_number=None)
Checkpoints with retries on retryable exceptions.

        :param amazon_kclpy.kcl.Checkpointer checkpointer: the checkpointer provided to either process_records
            or shutdown
        :param str or None sequence_number: the sequence number to checkpoint at.
        :param int or None sub_sequence_number: the sub sequence number to checkpoint at.
Checkpoints with retries on retryable exceptions.
Checkpoints with retries on retryable exceptions .
def checkpoint(self, checkpointer, sequence_number=None, sub_sequence_number=None):
        """
        Checkpoints with retries on retryable exceptions.

        :param amazon_kclpy.kcl.Checkpointer checkpointer: the checkpointer provided to either process_records
            or shutdown
        :param str or None sequence_number: the sequence number to checkpoint at.
        :param int or None sub_sequence_number: the sub sequence number to checkpoint at.
        """
        for n in range(0, self._CHECKPOINT_RETRIES):
            try:
                checkpointer.checkpoint(sequence_number, sub_sequence_number)
                return
            except kcl.CheckpointError as e:
                if 'ShutdownException' == e.value:
                    #
                    # A ShutdownException indicates that this record processor should be shutdown. This is due to
                    # some failover event, e.g. another MultiLangDaemon has taken the lease for this shard.
                    #
                    print('Encountered shutdown exception, skipping checkpoint')
                    return
                elif 'ThrottlingException' == e.value:
                    #
                    # A ThrottlingException indicates that one of our dependencies is is over burdened, e.g. too many
                    # dynamo writes. We will sleep temporarily to let it recover.
                    #
                    if self._CHECKPOINT_RETRIES - 1 == n:
                        sys.stderr.write('Failed to checkpoint after {n} attempts, giving up.\n'.format(n=n))
                        return
                    else:
                        print('Was throttled while checkpointing, will attempt again in {s} seconds'
                              .format(s=self._SLEEP_SECONDS))
                elif 'InvalidStateException' == e.value:
                    sys.stderr.write('MultiLangDaemon reported an invalid state while checkpointing.\n')
                else:  # Some other error
                    sys.stderr.write('Encountered an error while checkpointing, error was {e}.\n'.format(e=e))
            time.sleep(self._SLEEP_SECONDS)
def checkpoint ( self , checkpointer , sequence_number = None , sub_sequence_number = None ) : for n in range ( 0 , self . _CHECKPOINT_RETRIES ) : try : checkpointer . checkpoint ( sequence_number , sub_sequence_number ) return except kcl . CheckpointError as e : if 'ShutdownException' == e . value : # # A ShutdownException indicates that this record processor should be shutdown. This is due to # some failover event, e.g. another MultiLangDaemon has taken the lease for this shard. # print ( 'Encountered shutdown exception, skipping checkpoint' ) return elif 'ThrottlingException' == e . value : # # A ThrottlingException indicates that one of our dependencies is is over burdened, e.g. too many # dynamo writes. We will sleep temporarily to let it recover. # if self . _CHECKPOINT_RETRIES - 1 == n : sys . stderr . write ( 'Failed to checkpoint after {n} attempts, giving up.\n' . format ( n = n ) ) return else : print ( 'Was throttled while checkpointing, will attempt again in {s} seconds' . format ( s = self . _SLEEP_SECONDS ) ) elif 'InvalidStateException' == e . value : sys . stderr . write ( 'MultiLangDaemon reported an invalid state while checkpointing.\n' ) else : # Some other error sys . stderr . write ( 'Encountered an error while checkpointing, error was {e}.\n' . format ( e = e ) ) time . sleep ( self . _SLEEP_SECONDS )