Method ReadFromShard

pkg/acquisition/modules/kinesis/run.go:330–387

(ctx context.Context, out chan pipeline.Event, shardID string)

}

func (s *Source) ReadFromShard(ctx context.Context, out chan pipeline.Event, shardID string) error {
	logger := s.logger.WithField("shard", shardID)
	logger.Debugf("Starting to read shard")

	sharIt, err := s.kClient.GetShardIterator(ctx,
		&kinesis.GetShardIteratorInput{
			ShardId:           aws.String(shardID),
			StreamName:        &s.Config.StreamName,
			ShardIteratorType: kinTypes.ShardIteratorTypeLatest,
		})
	if err != nil {
		logger.Errorf("Cannot get shard iterator: %s", err)
		return fmt.Errorf("cannot get shard iterator: %w", err)
	}

	it := sharIt.ShardIterator
	// AWS recommends to wait for a second between calls to GetRecords for a given shard
	ticker := time.NewTicker(time.Second)

	for {
		select {
		case <-ticker.C:
			records, err := s.kClient.GetRecords(ctx, &kinesis.GetRecordsInput{ShardIterator: it})

			var throughputErr *kinTypes.ProvisionedThroughputExceededException
			if errors.As(err, &throughputErr) {
				logger.Warn("Provisioned throughput exceeded")
				// TODO: implement exponential backoff
				continue
			}

			var expiredIteratorErr *kinTypes.ExpiredIteratorException
			if errors.As(err, &expiredIteratorErr) {
				logger.Warn("Expired iterator")
				continue
			}

			if err != nil {
				logger.Error("Cannot get records")
				return fmt.Errorf("cannot get records: %w", err)
			}

			it = records.NextShardIterator

			s.ParseAndPushRecords(records.Records, out, logger, shardID)

			if it == nil {
				logger.Warnf("Shard has been closed")
				return nil
			}
		case <-s.shardReaderTomb.Dying():
			logger.Infof("shardReaderTomb is dying, exiting ReadFromShard")
			ticker.Stop()

			return nil
		}
	}
}