title: AWS Kinesis - how to resume consuming from last checkpoint
body: <p>I'm converting a Kafka consumer to an AWS Kinesis consumer, using the KCL (v2). In Kafka, offsets are used to help a consumer keep track of its most recently-consumed message. If my Kafka app dies, it will use the offset to consume from where it left off when it restarts.</p>
  
  <p>However this isn't the same in Kinesis. I can set <code>kinesisClientLibConfiguration.withInitialPositionInStream(...)</code> but the only arguments for that are <code>TRIM_HORIZON</code>, <code>LATEST</code> or <code>AT_TIMESTAMP</code>. If my Kinesis app died, it would not know where to resume consuming from when it restarts.</p>
  
  <p>My KCL consumer is very simple. The <code>main()</code> method looks like:</p>
  
  <pre><code>KinesisClientLibConfiguration config = new KinesisClientLibConfiguration("benTestApp",
              "testStream", new DefaultAWSCredentialsProviderChain(), UUID.randomUUID().toString());
  config.withInitialPositionInStream(InitialPositionInStream.TRIM_HORIZON);
  
  Worker worker = new Worker.Builder()
              .recordProcessorFactory(new KCLRecordProcessorFactory())
              .config(config)
              .build();
  </code></pre>
  
  <p>and the <code>RecordProcessor</code> is a simple implementation:</p>
  
  <pre><code>@Override
  public void initialize(InitializationInput initializationInput) {
      LOGGER.info("Initializing record processor for shard: {}", initializationInput.getShardId());
  }
  
  @Override
  public void processRecords(ProcessRecordsInput processRecordsInput) {
      List&lt;Record&gt; records = processRecordsInput.getRecords();
      LOGGER.info("Retrieved {} records", records.size());
      records.forEach(r -&gt; LOGGER.info("Record: {}", StandardCharsets.UTF_8.decode(r.getData())));
  }
  
  @Override
  public void shutdown(ShutdownInput shutdownInput) {
      LOGGER.info("Shutting down input");
  }
  </code></pre>
  
  <p>If I check the corresponding DynamoDB table, the value of <code>checkpoint</code> is set as <code>TRIM_HORIZON</code>, and does not get updated with sequenceIds as records are consumed.</p>
  
  <p>What's the solution here to ensure I consume every message?</p>