修复将记录添加到 Kinesis Stream 时的 AWS Lambda 超时问题

排查 AWS Lambda for Kinesis 流处理中的超时问题

方法 1：使用具有优化重试和自定义错误处理功能的 AWS 开发工具包的 JavaScript 解决方案

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();
}

替代 Lambda 配置可提高网络调用的弹性

方法 2：具有可调整超时和重试机制的增强型 JavaScript 解决方案

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;
  }
}

针对不同环境对 Lambda 函数进行单元测试

方法 3：使用 Jest 进行 JavaScript 单元测试来验证 Kinesis 流集成

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');
  });
});

有关 AWS Lambda 和 Kinesis 超时问题的常见问题解答

1. 是什么原因造成的 ETIMEDOUT 连接到 Kinesis 时 AWS Lambda 出现错误？
2. 这些错误通常在 Lambda 连接到 Kinesis 所需时间过长时发生，通常是由于网络问题、连接超时设置或 Kinesis 流上的高流量造成的。
3. 怎样才能调整 connectTimeout 有助于防止超时错误？
4. 设置更高 connectTimeout 允许 Lambda 等待响应的时间更长，这在网络延迟较高或数据流量较大的情况下很有帮助。
5. 为什么是 TextEncoder().encode() 此 Lambda 函数中使用的方法？
6. Kinesis 要求数据采用二进制格式。这 TextEncoder().encode() 方法将 JSON 数据转换为所需的格式，使其能够被 Kinesis 正确处理。
7. 在 Kinesis 中使用动态分区键有何重要性？
8. 动态密钥在分片之间更均匀地分配记录，避免瓶颈并减少“热分片”的机会，“热分片”可能导致流问题。
9. 单元测试可以模拟超时错误吗？
10. 是的，通过嘲笑 KinesisClient 在测试环境中，您可以模拟超时错误以验证 Lambda 函数中的错误处理是否正常工作。
11. 为什么这样做 Promise.allSettled() 和 Promise.all() 表现不同？
12. Promise.allSettled() 无论结果如何，都会等待所有承诺，这使其成为处理部分失败的多个请求的理想选择，这与 Promise.all()，在第一次失败时停止。
13. Lambda 中的重试次数是否有限制？
14. 是的， maxRetries 设置控制 Lambda 重试失败请求的次数，这可以减少网络负载，但应谨慎设置。
15. 区域选择在减少超时方面发挥什么作用？
16. 选择距离数据源较近的区域可以减少延迟，使与 Kinesis 的连接更快并且不易出现超时错误。
17. 怎么样 Promise.allSettled() 协助处理 Lambda 错误？
18. 它允许函数单独处理每个 Promise 结果，因此如果一个请求失败，其余请求仍会继续。这种方法有利于管理批量记录处理。
19. Lambda 可以处理流数据的部分成功吗？
20. 是的，使用 Promise.allSettled() 记录失败的记录使 Lambda 能够继续处理，即使某些记录遇到错误。

使用 AWS Lambda 和 Kinesis 克服常见挑战

对 Lambda 和 Kinesis 超时进行有效故障排除需要分析连接和配置问题。调整设置如 连接超时 和 最大重试次数以及周到的分区密钥管理，有助于维护可靠的连接并防止常见的超时。通过这些策略，处理高吞吐量数据流变得更加顺畅。 🚀

通过了解如何处理错误和优化配置，开发人员可以解决发布到 Kinesis 的 Lambda 函数中持续出现的 ETIMEDOUT 错误。遵循网络设置、编码和分区的最佳实践有助于提高数据管道的弹性和效率，确保更少的中断和更好的性能。