Akka Streams调用DynamoDB无业务栈异常排查及配置咨询

我们有一个基于Akka Streams的Scala应用，从Kafka Topic消费消息并转换为DynamoDB请求执行。偶尔会抛出含超时的SdkClientException，但堆栈跟踪仅指向库代码，无业务源码调用链。现咨询异常根源排查方向及Akka异常栈配置方法，附核心代码示例：

核心请求代码：

def makeRequest[In <: DynamoDbRequest, Out <: DynamoDbResponse](
    client: DynamoDbAsyncClient,
    retries: DynamoDBRetries,
    request: In
)(implicit
    dbOp: DynamoDbOp[In, Out],
    system: ActorSystem
): Future[Out] = {
  implicit val implicitClient: DynamoDbAsyncClient = client
  val source = RestartSource.onFailuresWithBackoff(
    RestartSettings(retries.minBackoff, retries.maxBackoff, retries.randomFactor)
  ) { () =>
    Source.single(request).via(DynamoDb.flow(1)).map(Success.apply).recover {
      // some error handling
    }
  }
  source.map(_.get).runWith(Sink.head)
}

链式调用代码：

makeRequest(
  client,
  config.retries,
  req
).flatMap { result => makeRequest(
  client,
  config.retries,
  anotherRequest
) }

异常堆栈信息：

java.util.concurrent.CompletionException: software.amazon.awssdk.core.exception.SdkClientException: Unable to execute HTTP request: Response entity was not subscribed after 1 second. Make sure to read the response `entity` body or call `entity.discardBytes()` on it -- in case you deal with `HttpResponse`, use the shortcut `response.discardEntityBytes()`. POST / Default(104 bytes) -> 200 OK Default(13512 bytes)
    at software.amazon.awssdk.utils.CompletableFutureUtils.errorAsCompletionException(CompletableFutureUtils.java:65)
    ...

异常根源排查方向 #

1. DynamoDB响应处理问题 #

异常提示明确指出Response entity was not subscribed after 1 second，说明AWS SDK的HTTP客户端返回的响应实体未被及时订阅或丢弃：

• 检查DynamoDb.flow实现：确认是否正确处理了DynamoDB响应，有没有遗漏响应体的消费或丢弃操作。比如响应体较大时，流处理未及时订阅会触发超时。
• 排查链式调用的阻塞风险：两次makeRequest用flatMap串联，若第一个请求的响应处理耗时过长，可能占用客户端资源，影响后续响应体的处理。
• 客户端配置检查：查看DynamoDbAsyncClient的HTTP客户端配置，比如响应超时时间是否过短、连接池是否耗尽。可通过SdkHttpClient的httpClientBuilder调整responseTimeout等参数。

2. Akka Streams背压与资源问题 #

• 检查RestartSource重试逻辑：当前RestartSource.onFailuresWithBackoff的配置是否合理？重试过于频繁可能导致客户端连接池过载，引发响应处理超时。
• 流并行度配置：DynamoDb.flow(1)设置并行度为1，高负载下可能成为瓶颈。可适当调高并行度，同时确保DynamoDB客户端连接池配置匹配。
• 下游消费速度：若Sink.head之后的业务逻辑处理缓慢，会触发流的背压传递到上游，导致响应体无法及时处理，进而触发超时。

Akka异常栈配置方法 #

要让异常堆栈包含业务代码调用链，需调整Akka的日志和异常配置：

1. 开启完整堆栈跟踪 #

在application.conf中添加以下配置：

akka.actor.default-dispatcher {
  fork-join-exception-handler {
    mode = full-stack-trace
  }
}
akka.stream.materializer {
  dispatcher = akka.actor.default-dispatcher
  logging-filter = akka.event.slf4j.Slf4jLoggingFilter
}

该配置会让Akka处理流异常时保留完整调用栈，包括业务代码链路。

2. 优化Future异常捕获 #

在makeRequest方法中，避免recover块吞掉关键堆栈信息，可修改为：

.recover {
  case e: SdkClientException =>
    // 保留原始异常堆栈，不丢失调用链
    Failure(new RuntimeException("DynamoDB请求失败", e))
  // 其他异常处理逻辑
}

3. 调整日志配置 #

在日志配置文件（如logback.xml）中，设置相关日志级别为DEBUG，确保输出完整堆栈：

<logger name="akka.stream" level="DEBUG"/>
<logger name="software.amazon.awssdk" level="DEBUG"/>

这能获取更多流处理和AWS SDK内部的细节日志，辅助定位问题。

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