getRecords

import {
  AWSConfig,
  KinesisClient,
} from 'https://jslib.k6.io/aws/0.14.0/kinesis.js';

const awsConfig = new AWSConfig({
  region: __ENV.AWS_REGION,
  accessKeyId: __ENV.AWS_ACCESS_KEY_ID,
  secretAccessKey: __ENV.AWS_SECRET_ACCESS_KEY,
});

const kinesis = new KinesisClient(awsConfig);

export default async function () {
  const streamName = 'my-test-stream';

  // First, get the shards for the stream
  const shards = await kinesis.listShards(streamName);

  if (shards.shards.length > 0) {
    const shardId = shards.shards[0].shardId;

    // Get a shard iterator for the first shard
    const shardIteratorResponse = await kinesis.getShardIterator(
      streamName,
      shardId,
      'TRIM_HORIZON'
    );

    const shardIterator = shardIteratorResponse.shardIterator;

    // Get records from the shard
    const recordsResponse = await kinesis.getRecords(shardIterator, { limit: 10 });

    console.log('Records retrieved:', recordsResponse.records.length);
    console.log('Milliseconds behind latest:', recordsResponse.millisBehindLatest);

    // Process the records
    recordsResponse.records.forEach((record, index) => {
      console.log(`Record ${index}:`);
      console.log('  Sequence number:', record.sequenceNumber);
      console.log('  Partition key:', record.partitionKey);
      console.log('  Data:', record.data);
      console.log('  Arrival timestamp:', record.approximateArrivalTimestamp);

      // Parse JSON data if applicable
      try {
        const jsonData = JSON.parse(record.data);
        console.log('  Parsed data:', jsonData);
      } catch (e) {
        console.log('  Data is not JSON');
      }
    });

    // Continue reading with the next shard iterator
    if (recordsResponse.nextShardIterator) {
      const nextBatch = await kinesis.getRecords(recordsResponse.nextShardIterator, { limit: 5 });
      console.log('Next batch size:', nextBatch.records.length);
    }
  }
}