diff --git a/src/main/java/com/googlesource/gerrit/plugins/kinesis/Configuration.java b/src/main/java/com/googlesource/gerrit/plugins/kinesis/Configuration.java
index ef7e698..9c04c09 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kinesis/Configuration.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kinesis/Configuration.java
@@ -38,6 +38,7 @@
   private static final Long DEFAULT_PUBLISH_SINGLE_REQUEST_TIMEOUT_MS = 6000L;
   private static final Long DEFAULT_PUBLISH_TIMEOUT_MS = 6000L;
   private static final Long DEFAULT_SHUTDOWN_TIMEOUT_MS = 20000L;
+  private static final Long DEFAULT_CHECKPOINT_INTERVAL_MS = 5 * 60000L; // 5 min
   private static final Level DEFAULT_AWS_LIB_LOG_LEVEL = Level.WARN;
   private static final Boolean DEFAULT_SEND_ASYNC = true;
 
@@ -52,6 +53,7 @@
   private final Long publishTimeoutMs;
   private final Long publishSingleRequestTimeoutMs;
   private final Long shutdownTimeoutMs;
+  private final Long checkpointIntervalMs;
   private final Level awsLibLogLevel;
   private final Boolean sendAsync;
 
@@ -98,6 +100,11 @@
             .map(Long::parseLong)
             .orElse(DEFAULT_SHUTDOWN_TIMEOUT_MS);
 
+    this.checkpointIntervalMs =
+        Optional.ofNullable(getStringParam(pluginConfig, "checkpointIntervalMs", null))
+            .map(Long::parseLong)
+            .orElse(DEFAULT_CHECKPOINT_INTERVAL_MS);
+
     this.awsLibLogLevel =
         Optional.ofNullable(getStringParam(pluginConfig, "awsLibLogLevel", null))
             .map(l -> Level.toLevel(l, DEFAULT_AWS_LIB_LOG_LEVEL))
@@ -172,6 +179,10 @@
     return shutdownTimeoutMs;
   }
 
+  public Long getCheckpointIntervalMs() {
+    return checkpointIntervalMs;
+  }
+
   public Level getAwsLibLogLevel() {
     return awsLibLogLevel;
   }
diff --git a/src/main/java/com/googlesource/gerrit/plugins/kinesis/KinesisRecordProcessor.java b/src/main/java/com/googlesource/gerrit/plugins/kinesis/KinesisRecordProcessor.java
index 2786b82..29c7f60 100644
--- a/src/main/java/com/googlesource/gerrit/plugins/kinesis/KinesisRecordProcessor.java
+++ b/src/main/java/com/googlesource/gerrit/plugins/kinesis/KinesisRecordProcessor.java
@@ -24,11 +24,13 @@
 import java.util.function.Consumer;
 import software.amazon.kinesis.exceptions.InvalidStateException;
 import software.amazon.kinesis.exceptions.ShutdownException;
+import software.amazon.kinesis.exceptions.ThrottlingException;
 import software.amazon.kinesis.lifecycle.events.InitializationInput;
 import software.amazon.kinesis.lifecycle.events.LeaseLostInput;
 import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput;
 import software.amazon.kinesis.lifecycle.events.ShardEndedInput;
 import software.amazon.kinesis.lifecycle.events.ShutdownRequestedInput;
+import software.amazon.kinesis.processor.RecordProcessorCheckpointer;
 import software.amazon.kinesis.processor.ShardRecordProcessor;
 
 class KinesisRecordProcessor implements ShardRecordProcessor {
@@ -40,21 +42,29 @@
   private final Consumer<Event> recordProcessor;
   private final OneOffRequestContext oneOffCtx;
   private final EventDeserializer eventDeserializer;
+  private final Configuration configuration;
+
+  private long nextCheckpointTimeInMillis;
+  private String kinesisShardId;
 
   @Inject
   KinesisRecordProcessor(
       @Assisted Consumer<Event> recordProcessor,
       OneOffRequestContext oneOffCtx,
-      EventDeserializer eventDeserializer) {
+      EventDeserializer eventDeserializer,
+      Configuration configuration) {
     this.recordProcessor = recordProcessor;
     this.oneOffCtx = oneOffCtx;
     this.eventDeserializer = eventDeserializer;
+    this.configuration = configuration;
   }
 
   @Override
   public void initialize(InitializationInput initializationInput) {
+    kinesisShardId = initializationInput.shardId();
     logger.atInfo().log(
         "Initializing @ Sequence: %s", initializationInput.extendedSequenceNumber());
+    setNextCheckpointTime();
   }
 
   @Override
@@ -79,11 +89,21 @@
                   logger.atSevere().withCause(e).log("Could not process event '%s'", jsonMessage);
                 }
               });
+
+      if (System.currentTimeMillis() >= nextCheckpointTimeInMillis) {
+        checkpoint(processRecordsInput.checkpointer());
+        setNextCheckpointTime();
+      }
     } catch (Throwable t) {
       logger.atSevere().withCause(t).log("Caught throwable while processing records. Aborting.");
     }
   }
 
+  private void setNextCheckpointTime() {
+    nextCheckpointTimeInMillis =
+        System.currentTimeMillis() + configuration.getCheckpointIntervalMs();
+  }
+
   @Override
   public void leaseLost(LeaseLostInput leaseLostInput) {
     logger.atInfo().log("Lost lease, so terminating.");
@@ -91,22 +111,28 @@
 
   @Override
   public void shardEnded(ShardEndedInput shardEndedInput) {
-    try {
-      logger.atInfo().log("Reached shard end checkpointing.");
-      shardEndedInput.checkpointer().checkpoint();
-    } catch (ShutdownException | InvalidStateException e) {
-      logger.atSevere().withCause(e).log("Exception while checkpointing at shard end. Giving up.");
-    }
+    logger.atInfo().log("Reached shard end checkpointing.");
+    checkpoint(shardEndedInput.checkpointer());
   }
 
   @Override
   public void shutdownRequested(ShutdownRequestedInput shutdownRequestedInput) {
+    logger.atInfo().log("Scheduler is shutting down, checkpointing.");
+    checkpoint(shutdownRequestedInput.checkpointer());
+  }
+
+  private void checkpoint(RecordProcessorCheckpointer checkpointer) {
+    logger.atInfo().log("Checkpointing shard: " + kinesisShardId);
     try {
-      logger.atInfo().log("Scheduler is shutting down, checkpointing.");
-      shutdownRequestedInput.checkpointer().checkpoint();
-    } catch (ShutdownException | InvalidStateException e) {
+      checkpointer.checkpoint();
+    } catch (ShutdownException se) {
+      logger.atInfo().log("Caught shutdown exception, skipping checkpoint.", se);
+    } catch (ThrottlingException e) {
+      logger.atSevere().withCause(e).log("Caught throttling exception, skipping checkpoint.", e);
+    } catch (InvalidStateException e) {
       logger.atSevere().withCause(e).log(
-          "Exception while checkpointing at requested shutdown. Giving up.");
+          "Cannot save checkpoint to the DynamoDB table used by the Amazon Kinesis Client Library.",
+          e);
     }
   }
 }
diff --git a/src/main/resources/Documentation/Config.md b/src/main/resources/Documentation/Config.md
index 03e3650..004ce0b 100644
--- a/src/main/resources/Documentation/Config.md
+++ b/src/main/resources/Documentation/Config.md
@@ -81,6 +81,10 @@
   kinesis consumers.
   Default: 20000
 
+`plugin.events-aws-kinesis.checkpointIntervalMs`
+: Optional. The interval between checkpoints (milliseconds).
+Default: 300000 (5 minutes)
+
 `plugin.events-aws-kinesis.awsLibLogLevel`
 : Optional. Which level AWS libraries should log at.
   This plugin delegates most complex tasks associated to the production and
diff --git a/src/test/java/com/googlesource/gerrit/plugins/kinesis/KinesisRecordProcessorTest.java b/src/test/java/com/googlesource/gerrit/plugins/kinesis/KinesisRecordProcessorTest.java
index d488ab8..24027fc 100644
--- a/src/test/java/com/googlesource/gerrit/plugins/kinesis/KinesisRecordProcessorTest.java
+++ b/src/test/java/com/googlesource/gerrit/plugins/kinesis/KinesisRecordProcessorTest.java
@@ -38,11 +38,14 @@
 import org.mockito.ArgumentCaptor;
 import org.mockito.Captor;
 import org.mockito.Mock;
+import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
 import software.amazon.awssdk.core.SdkBytes;
 import software.amazon.awssdk.services.kinesis.model.Record;
+import software.amazon.kinesis.lifecycle.events.InitializationInput;
 import software.amazon.kinesis.lifecycle.events.ProcessRecordsInput;
 import software.amazon.kinesis.retrieval.KinesisClientRecord;
+import software.amazon.kinesis.retrieval.kpl.ExtendedSequenceNumber;
 
 @RunWith(MockitoJUnitRunner.class)
 public class KinesisRecordProcessorTest {
@@ -54,11 +57,41 @@
   @Captor ArgumentCaptor<Event> eventMessageCaptor;
   @Mock OneOffRequestContext oneOffCtx;
   @Mock ManualRequestContext requestContext;
+  @Mock Configuration configuration;
 
   @Before
   public void setup() {
     when(oneOffCtx.open()).thenReturn(requestContext);
-    objectUnderTest = new KinesisRecordProcessor(succeedingConsumer, oneOffCtx, eventDeserializer);
+    objectUnderTest =
+        new KinesisRecordProcessor(succeedingConsumer, oneOffCtx, eventDeserializer, configuration);
+  }
+
+  @Test
+  public void shouldNotCheckpointBeforeIntervalIsExpired() {
+    when(configuration.getCheckpointIntervalMs()).thenReturn(10000L);
+    Event event = new ProjectCreatedEvent();
+
+    initializeRecordProcessor();
+
+    ProcessRecordsInput kinesisInput = sampleMessage(gson.toJson(event));
+    ProcessRecordsInput processRecordsInputSpy = Mockito.spy(kinesisInput);
+    objectUnderTest.processRecords(processRecordsInputSpy);
+
+    verify(processRecordsInputSpy, never()).checkpointer();
+  }
+
+  @Test
+  public void shouldCheckpointAfterIntervalIsExpired() throws InterruptedException {
+    when(configuration.getCheckpointIntervalMs()).thenReturn(0L);
+    Event event = new ProjectCreatedEvent();
+
+    initializeRecordProcessor();
+
+    ProcessRecordsInput kinesisInput = sampleMessage(gson.toJson(event));
+    ProcessRecordsInput processRecordsInputSpy = Mockito.spy(kinesisInput);
+    objectUnderTest.processRecords(processRecordsInputSpy);
+
+    verify(processRecordsInputSpy).checkpointer();
   }
 
   @Test
@@ -153,4 +186,13 @@
             .build();
     return kinesisInput;
   }
+
+  private void initializeRecordProcessor() {
+    InitializationInput initializationInput =
+        InitializationInput.builder()
+            .shardId("shard-0000")
+            .extendedSequenceNumber(new ExtendedSequenceNumber("0000"))
+            .build();
+    objectUnderTest.initialize(initializationInput);
+  }
 }