ResetShardIterator added to rp interface (#2)

clientlibrary/interfaces/record-processor.go

@@ -61,6 +61,12 @@ type (
 		 */
 		ProcessRecords(processRecordsInput *ProcessRecordsInput)
 
+		// ResetShardIterator
+		/*
+		 * If true, resets the shard iterator to the last checkpointed sequence number for the shard.
+		 */
+		ResetShardIterator() bool
+
 		// Shutdown
 		/*
 		 * Invoked by the Amazon Kinesis Client Library to indicate it will no longer send data records to this

clientlibrary/worker/polling-shard-consumer.go

@@ -127,6 +127,17 @@ func (sc *PollingShardConsumer) getRecords() error {
 		getRecordsStartTime := time.Now()
 
 		log.Debugf("Trying to read %d record from iterator: %v", sc.kclConfig.MaxRecords, aws.ToString(shardIterator))
+
+		// check if ResetShardIterator returns true
+		if sc.recordProcessor.ResetShardIterator() {
+			// reset shard iterator
+			shardIterator, err = sc.getShardIterator()
+			if err != nil {
+				log.Errorf("Unable to get shard iterator for %s: %v", sc.shard.ID, err)
+				return err
+			}
+		}
+
 		getRecordsArgs := &kinesis.GetRecordsInput{
 			Limit:         aws.Int32(int32(sc.kclConfig.MaxRecords)),
 			ShardIterator: shardIterator,

test/record_processor_test.go

@@ -57,6 +57,10 @@ func (dd *dumpRecordProcessor) Initialize(input *kc.InitializationInput) {
 	dd.count = 0
 }
 
+func (dd *dumpRecordProcessor) ResetShardIterator() bool {
+	return false
+}
+
 func (dd *dumpRecordProcessor) ProcessRecords(input *kc.ProcessRecordsInput) {
 	dd.t.Log("Processing Records...")