Fixing AWS Lambda Timeout Issues When Adding Records to Kinesis Stream

Troubleshooting AWS Lambda Timeouts for Kinesis Data Streams

Imagine you're building a real-time data pipeline on AWS, with a setup that passes messages from SQS to a Lambda function, and ultimately to a Kinesis Data Stream. 📨 This flow works seamlessly in theory, but sometimes reality has other plans. Just when you’re about to relax, an ETIMEDOUT error crops up in your Lambda function logs.

Seeing this error can be frustrating, especially when you've verified permissions and tested the function multiple times. In fact, this intermittent ETIMEDOUT issue in the Kinesis stream usually happens unexpectedly, halting your progress. The Lambda might work perfectly after a redeployment but then fail again, seemingly without reason.

In situations like this, many developers have been stumped by cryptic messages like "Runtime.UnhandledPromiseRejection" and "ERR_HTTP2_STREAM_CANCEL." When your code relies on reliable and immediate data processing, these timeout issues can feel like a roadblock.

Here, we’ll go over what causes these timeouts, practical ways to handle them, and adjustments in your AWS configuration that may just be the key to stabilizing your stream. 🛠️ By the end, you’ll know how to troubleshoot and resolve ETIMEDOUT errors and keep your Lambda and Kinesis flow running smoothly.

Enhancing AWS Lambda Reliability with Kinesis Stream

The provided JavaScript scripts are designed to create an efficient AWS Lambda function that retrieves messages from an SQS queue and then publishes them to an Amazon Kinesis Data Stream. The core of this solution lies in the Lambda function’s ability to handle messages asynchronously while addressing connectivity issues that frequently result in ETIMEDOUT errors. One key part of the script is the initialization of the KinesisClient, which configures essential properties like region, retry count, and connection timeout. These configurations are critical in a cloud setup, as they control the responsiveness of the application and how long it will attempt to connect before timing out. By setting a higher connectTimeout or adjusting retry attempts, we can help the function handle network delays more effectively.

Within the Lambda handler, the script leverages Promise.allSettled(), an invaluable tool when processing multiple asynchronous requests. When multiple records are processed at once, it’s essential to ensure each one completes, whether successfully or with an error. Promise.allSettled() ensures that the function doesn't stop processing if one request fails; instead, it logs each result individually. This approach is especially useful in situations where network connectivity might be unpredictable. For example, if one record fails due to a network issue but others succeed, the function can log the failed records separately, allowing developers to isolate problem instances instead of failing the entire batch of messages. 🛠️

The processEvent function within the script is modular and handles the main data transformation and sending process. This function takes in the SQS message, parses it, and encodes it into the byte format that Kinesis requires. Here, the TextEncoder().encode() method is critical as Kinesis accepts only binary data; JSON must be converted to a compatible format. This part of the function ensures that the Lambda sends data correctly, reducing the likelihood of errors arising from mismatched data formats. The function also uses a custom partition key generator function, which distributes records across the Kinesis stream’s shards. By using dynamic partition keys (such as random keys), the script minimizes the chances of hitting the same shard repeatedly, which can prevent “hot shards” that lead to bottlenecks.

Lastly, to ensure this setup functions correctly across various scenarios, the scripts incorporate unit tests using Jest. Unit tests make it possible to simulate the Kinesis client’s behavior without needing live AWS resources, offering a reliable way to test the Lambda’s ability to handle timeouts or data conversion issues in a controlled environment. For instance, if the Kinesis client is unable to connect, Jest mocks can simulate a timeout error, verifying that the error handling within processEvent works as intended. This strategy allows for robust validation, ensuring that the Lambda is reliable across multiple network conditions. 🧪 With these elements combined, the Lambda function can handle data from SQS to Kinesis efficiently while minimizing timeouts and other common streaming errors.

import { KinesisClient, PutRecordCommand } from "@aws-sdk/client-kinesis";
const KINESIS_CLIENT = new KinesisClient({
  region: 'us-west-2',
  maxRetries: 3,
  requestHandler: {
    connectionTimeout: 5000, // Lower timeout for quicker retries
  },
});
export const handler = async (event) => {
  const promises = event.Records.map(record => processEvent(record.body));
  const results = await Promise.allSettled(promises);
  results.forEach(result => {
    if (result.status !== "fulfilled") {
      console.error("Failed:", result.reason);
    }
  });
};
async function processEvent(body) {
  const data = JSON.parse(body);
  const partitionKey = generatePartitionKey(data);
  const encodedData = new TextEncoder().encode(JSON.stringify(data));
  try {
    await KINESIS_CLIENT.send(new PutRecordCommand({
      Data: encodedData,
      StreamName: 'InputEventStream',
      PartitionKey: partitionKey
    }));
  } catch (error) {
    console.error("Error putting record:", error);
    throw error;
  }
}
function generatePartitionKey(data) {
  return data.userId ? data.userId.toString() : Date.now().toString();
}

import { KinesisClient, PutRecordCommand } from "@aws-sdk/client-kinesis";
const KINESIS_CLIENT = new KinesisClient({
  region: 'us-west-2',
  maxRetries: 5,
  httpOptions: {
    connectTimeout: 15000, // Extended timeout
    timeout: 20000 // Total request timeout
  }
});
export const handler = async (event) => {
  const results = await Promise.allSettled(event.Records.map(async (record) => {
    await processEvent(record.body);
  }));
  results.forEach((result) => {
    if (result.status !== "fulfilled") {
      console.log("Unsuccessful attempt:", result.reason);
    }
  });
};
async function processEvent(body) {
  const parsedData = JSON.parse(body);
  const partitionKey = `pk-${Math.random()}`;
  try {
    await KINESIS_CLIENT.send(new PutRecordCommand({
      StreamName: "InputEventStream",
      Data: new TextEncoder().encode(JSON.stringify(parsedData)),
      PartitionKey: partitionKey
    }));
  } catch (err) {
    if (err.name === "TimeoutError") {
      console.warn("Retry on timeout:", err);
    }
    throw err;
  }
}

import { handler, processEvent } from './your-lambda-file.js';
import { KinesisClient } from "@aws-sdk/client-kinesis";
jest.mock("@aws-sdk/client-kinesis");
describe('Lambda Handler and Kinesis Integration', () => {
  it('should call processEvent for each record in the event', async () => {
    const mockEvent = {
      Records: [{ body: '{"userId": 1, "data": "test"}' }]
    };
    await handler(mockEvent);
    expect(KinesisClient.prototype.send).toHaveBeenCalledTimes(1);
  });
  it('should handle timeout errors gracefully', async () => {
    KinesisClient.prototype.send.mockRejectedValueOnce(new Error('TimeoutError'));
    await expect(processEvent('{"userId": 2}')).rejects.toThrow('TimeoutError');
  });
});

Understanding Timeout Errors in AWS Lambda-Kinesis Integrations

Timeout errors like ETIMEDOUT in AWS Lambda functions can often be frustrating, especially in integrations involving data streaming with Amazon Kinesis. In most cases, these errors occur due to the Lambda function exceeding network connection time limits, typically during a KinesisClient request. The default settings in Lambda might not always accommodate these kinds of network requests, particularly when dealing with high-throughput streams or large amounts of data. For instance, adjusting the connectTimeout or maxRetries configurations can help mitigate this issue, allowing the Lambda more time to attempt a successful connection to Kinesis. This kind of optimization is often necessary in scenarios with variable network latency or under high demand. 🛠️

Another key aspect in reducing timeout errors is managing data encoding and partitioning effectively. AWS Kinesis requires data in binary format, which can be achieved through TextEncoder().encode(). This transformation ensures compatibility and streamlining of data transfer to Kinesis. Additionally, thoughtful partition key management is crucial. Using a consistent or dynamically generated partition key helps distribute data evenly across Kinesis shards, avoiding "hot shards," which are shards receiving a disproportionate number of records. In high-frequency streaming scenarios, dynamic keys can prevent bottlenecks and reduce the probability of connectivity issues, particularly useful when handling large datasets.

To troubleshoot and improve the reliability of these Lambda-Kinesis interactions, adding unit tests is essential. Unit tests allow you to simulate potential network issues, validate data encoding, and ensure that the function can handle retries correctly. For example, by mocking KinesisClient in unit tests, you can simulate a range of responses from Kinesis, such as timeout errors or success cases, which helps in fine-tuning error handling and connection management within the Lambda code. Testing for such error cases in development can lead to a more resilient deployment, reducing the likelihood of timeouts in production and making it easier to identify weak points in your configuration.