AWS Greengrass无法向AWS Kinesis发送数据问题排查求助

• StreamManagerClient实例频繁创建/销毁：如果初始代码每次处理MQTT消息时都新建StreamManagerClient并在写完后立即关闭，会导致客户端还没完成数据上传到Stream Manager服务就被终止，数据滞留在本地缓存无法同步到Kinesis。临时方案里的单例客户端避免了这个问题，而1秒延迟只是碰巧让部分数据赶上了提交窗口。
• 批量提交阈值未触发：Greengrass Stream Manager默认采用批量上传策略（比如累计到一定数据量或等待固定时间窗口）。如果单条MQTT数据量小、消息频率低，可能一直达不到触发条件，导致数据一直留在Stream Manager的本地缓存里，没有同步到Kinesis。
• 未等待异步提交完成：写入Stream Manager的操作是异步的，如果写完后直接关闭客户端或退出当前处理逻辑，会打断数据上传的异步流程，导致数据丢失。

1. 复用StreamManagerClient单例
   初始化阶段创建一次StreamManagerClient实例，全局复用，不要在每条MQTT消息处理逻辑里创建或关闭客户端。示例：

   # 全局初始化，仅执行一次
   stream_manager_client = StreamManagerClient()
   
   def on_mqtt_message_received(message):
       # 处理MQTT数据
       data = message.payload
       # 复用已有客户端写入流
       stream_manager_client.put_message("MyKinesisStream", data)
   

   StreamManagerClient是线程安全的，完全支持多线程场景下复用。

2. 调整Stream Manager的提交策略
   创建流时（动态创建场景），配置合适的批量参数，平衡实时性和性能：

   from awsiot.greengrasscore.streammanager import (
       StreamManagerClient,
       MessageStreamDefinition,
       KinesisStreamConfig,
       StrategyOnFull,
   )
   
   client = StreamManagerClient()
   # 创建流时配置立即提交或调整批量阈值
   client.create_message_stream(
       MessageStreamDefinition(
           name="MyKinesisStream",
           kinesis_stream_config=KinesisStreamConfig(
               stream_name="MyKinesisStream",
               # 启用写完立即提交（适合小数据量低频率场景）
               flush_on_write=True,
               # 或调整批量大小和时间窗口
               batch_size=1,
               batch_interval_milliseconds=500
           ),
           strategy_on_full=StrategyOnFull.OverwriteOldestData
       )
   )
   

   若为静态流配置，直接在Greengrass组件配置中修改对应批量参数即可。

3. 显式等待数据提交完成
   在写入消息后，调用flush方法等待Stream Manager完成数据上传到Kinesis：

   stream_manager_client.put_message("MyKinesisStream", data)
   # 等待当前流的所有待提交数据完成上传
   stream_manager_client.flush("MyKinesisStream")
   

   该操作会阻塞直到数据提交完成，确保数据不会因客户端关闭或逻辑退出而丢失。

4. 验证Stream Manager权限与配置

   • 确认Greengrass核心角色拥有kinesis:PutRecord和kinesis:PutRecords权限，目标Kinesis流名称与配置完全一致（大小写敏感）。
   • 查看Stream Manager日志（路径：/greengrass/v2/logs/aws.greengrass.StreamManager.log），检查是否存在Kinesis上传失败的错误信息，比如权限不足、流不存在等。

内容的提问来源于stack exchange，提问作者ForestG