Fix missing init position with AT_TIMESTAMP (#44)

AT_TIMESTAMP start from the record at or after the specified
server-side Timestamp. However, the implementation was
missing. The bug was not notices until recently because most
of users never use this feature.

Signed-off-by: Tao Jiang <taoj@vmware.com>

clientlibrary/worker/shard-consumer.go

@@ -95,13 +95,26 @@ func (sc *ShardConsumer) getShardIterator(shard *par.ShardStatus) (*string, erro
 	// If there isn't any checkpoint for the shard, use the configuration value.
 	if shard.Checkpoint == "" {
 		initPos := sc.kclConfig.InitialPositionInStream
+		shardIteratorType := config.InitalPositionInStreamToShardIteratorType(initPos)
 		log.Debugf("No checkpoint recorded for shard: %v, starting with: %v", shard.ID,
-			aws.StringValue(config.InitalPositionInStreamToShardIteratorType(initPos)))
+			aws.StringValue(shardIteratorType))
-		shardIterArgs := &kinesis.GetShardIteratorInput{
+
-			ShardId:           &shard.ID,
+		var shardIterArgs *kinesis.GetShardIteratorInput
-			ShardIteratorType: config.InitalPositionInStreamToShardIteratorType(initPos),
+		if initPos == config.AT_TIMESTAMP {
-			StreamName:        &sc.streamName,
+			shardIterArgs = &kinesis.GetShardIteratorInput{
+				ShardId:           &shard.ID,
+				ShardIteratorType: shardIteratorType,
+				Timestamp:         sc.kclConfig.InitialPositionInStreamExtended.Timestamp,
+				StreamName:        &sc.streamName,
+			}
+		} else {
+			shardIterArgs = &kinesis.GetShardIteratorInput{
+				ShardId:           &shard.ID,
+				ShardIteratorType: shardIteratorType,
+				StreamName:        &sc.streamName,
+			}
 		}
+
 		iterResp, err := sc.kc.GetShardIterator(shardIterArgs)
 		if err != nil {
 			return nil, err

test/worker_test.go

@@ -81,6 +81,30 @@ func TestWorker(t *testing.T) {
 	runTest(kclConfig, false, t)
 }
 
+func TestWorkerWithTimestamp(t *testing.T) {
+	// In order to have precise control over logging. Use logger with config
+	config := logger.Configuration{
+		EnableConsole:     true,
+		ConsoleLevel:      logger.Debug,
+		ConsoleJSONFormat: false,
+	}
+	// Use logrus logger
+	log := logger.NewLogrusLoggerWithConfig(config)
+
+	ts := time.Now().Add(time.Second * 5)
+	kclConfig := cfg.NewKinesisClientLibConfig("appName", streamName, regionName, workerID).
+		WithTimestampAtInitialPositionInStream(&ts).
+		WithMaxRecords(10).
+		WithMaxLeasesForWorker(1).
+		WithShardSyncIntervalMillis(5000).
+		WithFailoverTimeMillis(300000).
+		WithMetricsBufferTimeMillis(10000).
+		WithMetricsMaxQueueSize(20).
+		WithLogger(log)
+
+	runTest(kclConfig, false, t)
+}
+
 func TestWorkerWithSigInt(t *testing.T) {
 	// At miminal. use standard zap logger
 	//zapLogger, err := zap.NewProduction()