Akka Stream应用中HTTP连接意外关闭问题求助

问题背景 #

在基于Akka Stream(Scala)的应用中，流程为PubSub消息消费 → BigQuery查询 → Kafka消息生产 → PubSub消息确认，但间歇性出现以下错误导致流终止：

Runtime exception encountered: [{}], stopping stream 
akka.http.impl.engine.client.OutgoingConnectionBlueprint$UnexpectedConnectionClosureException: The http server closed the connection unexpectedly before delivering responses for 1 outstanding requests

已尝试调整超时、连接池参数，但问题仍存在；单独设置BigQuery idle-timeout会触发BigQuery自身的超时错误。

针对性解决方案 #

1. 修正Akka HTTP客户端配置的优先级 #

当前配置存在重复（如akka.http.idle-timeout与akka.http.client.idle-timeout），需明确Alpakka的GCP客户端使用的是客户端连接池的配置，而非服务器端配置：

akka {
  http {
    host-connection-pool {
      # 连接池级别的空闲超时，匹配GCP LB的超时（通常默认是60s，建议设为55s避免提前关闭）
      idle-timeout = 55s
      max-open-requests = 128
      # 允许重试意外关闭的连接
      max-retries = 3
    }
    client {
      request-timeout = 180s
      connecting-timeout = 30s
      # 请求级别的重试次数
      request-retries = 2
    }
  }
}

注：GCP负载均衡器默认会关闭超过60s的空闲连接，将idle-timeout设为略低于60s可以避免被主动断开。

2. 降低BigQuery查询的并行度 #

当前Kafka Producer的parallelism=10000过高，可能导致BigQuery请求并发超限，触发GCP侧的连接限流。建议将BigQuery查询的并行度调整为合理值（如4-8）：

// 原代码可能用了过高的并行度，改为：
GooglePubSub.source(subscriptionSettings)
  .mapAsync(4) { pubSubMsg => // 控制BigQuery查询的并发数
    BigQuery.query(QueryRequest(...))
      .map(result => (pubSubMsg, result))
  }

3. 为流添加容错处理 #

不要让单个请求的错误终止整个流，使用Akka Stream的监督策略或错误恢复逻辑：

// 定义监督策略，对连接意外关闭的错误进行恢复
val decider: Supervision.Decider = {
  case _: UnexpectedConnectionClosureException => Supervision.Resume // 跳过错误元素，继续流
  case ex => 
    println(s"不可恢复错误: ${ex.getMessage}")
    Supervision.Stop
}

// 在流中应用监督策略
val stream = GooglePubSub.source(subscriptionSettings)
  .mapAsync(4) { pubSubMsg =>
    BigQuery.query(QueryRequest(...))
      .recover {
        case ex: UnexpectedConnectionClosureException =>
          // 可选：记录日志，或者返回空结果
          println(s"BigQuery查询连接中断: ${ex.getMessage}")
          None
      }
  }
  .filter(_.isDefined)
  .map(_.get)
  .to(SendProducer.sink(kafkaProducerSettings))
  .withAttributes(ActorAttributes.supervisionStrategy(decider))

4. 检查GCP服务配额与监控 #

• 登录GCP控制台，查看BigQuery的请求速率、并发连接数指标，确认是否触发了配额限制；
• 若存在超限，可申请提高配额，或调整流的请求速率（如添加throttle算子）：

GooglePubSub.source(subscriptionSettings)
  .throttle(100, 1.second, 100, ThrottleMode.Shaping) // 限制每秒处理100条消息
  .mapAsync(4) { ... }

5. 升级Alpakka版本 #

检查当前使用的alpakka-google-cloud版本，若为较旧版本，可能存在HTTP连接池的已知bug。升级到最新稳定版本（如3.0.0+），可以修复部分连接管理的问题。

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