private void checkpoint ( IRecordProcessorCheckpointer checkpointer ) { LOG . info ( "Checkpointing shard " + kinesisShardId ) ; for ( int i = 0 ; i < NUM_RETRIES ; i ++ ) { try { checkpointer . checkpoint ( ) ; break ; } catch ( ShutdownException se ) { // Ignore checkpoint if the processor instance has been shutdown (fail over). LOG . info ( "Caught shutdown exception, skipping checkpoint." , se ) ; break ; } catch ( ThrottlingException e ) { // Backoff and re-attempt checkpoint upon transient failures if ( i >= ( NUM_RETRIES - 1 ) ) { LOG . error ( "Checkpoint failed after " + ( i + 1 ) + "attempts." , e ) ; break ; } else { LOG . info ( "Transient issue when checkpointing - attempt " + ( i + 1 ) + " of " + NUM_RETRIES , e ) ; } } catch ( InvalidStateException e ) { // This indicates an issue with the DynamoDB table (check for table, provisioned IOPS). LOG . error ( "Cannot save checkpoint to the DynamoDB table used by the Amazon Kinesis Client Library." , e ) ; break ; } try { Thread . sleep ( BACKOFF_TIME_IN_MILLIS ) ; } catch ( InterruptedException e ) { LOG . debug ( "Interrupted sleep" , e ) ; } } }
Checkpoint with retries .