如何解决Google DataFlow定时任务出现Connection reset by peer错误

• 优先核对Worker运行时网络配置
  手动运行作业时通常使用本地/测试网络权限，定时触发的DataFlow作业使用GCP VPC内的Worker节点资源，首先排查以下配置：
  • VPC是否配置了公网出口NAT网关，是否限制了外部接口的访问权限
  • 外部接口的IP白名单是否包含DataFlow Worker的出口IP段，若使用无服务器VPC访问需要确认NAT的静态IP是否已加入白名单
  • 定时作业指定的运行区域是否和手动测试区域一致，不同区域的出口IP段存在差异
• 修正请求重试与连接配置
  仅捕获通用异常无法解决底层网络断开问题，需要显式针对ConnectionResetError类异常配置重试，同时调整连接参数避免长连接复用问题，参考封装代码如下：

  import requests
  from requests.adapters import HTTPAdapter
  from urllib3.util.retry import Retry
  
  def make_request(url, data, timeout=10):
      session = requests.Session()
      # 配置重试策略，针对网络错误、服务端错误自动重试
      retry_strategy = Retry(
          total=3,
          backoff_factor=1,
          status_forcelist=[429, 500, 502, 503, 504],
          allowed_methods=["POST"],
          retry_on_connection_errors=True
      )
      adapter = HTTPAdapter(max_retries=retry_strategy)
      session.mount("https://", adapter)
      session.mount("http://", adapter)
      try:
          resp = session.post(
              url,
              json=data,
              timeout=timeout,
              headers={"Connection": "close"}
          )
          resp.raise_for_status()
          return resp.json()
      except (ConnectionResetError, ConnectionAbortedError, requests.exceptions.RequestException) as e:
          # 此处可新增错误日志落盘逻辑，便于后续排查
          raise e
  

• 确认Worker环境依赖版本
  本地升级requests版本不代表Worker运行时使用对应版本，需要在流水线的requirements.txt中显式指定依赖版本：

  requests==2.26.0
  urllib3>=1.26.0
  

  若使用DataFlow模板部署，需要在打包模板阶段同步更新依赖，避免使用系统默认的低版本依赖。
• 排查外部接口限流规则
  定时运行时的作业并发量、请求QPS通常高于手动测试场景，若超过外部接口的限流阈值，服务端会直接重置连接返回('Connection aborted.', ConnectionResetError(104, 'Connection reset by peer'))错误，可通过添加请求节流逻辑、调整作业并发数适配接口限流规则。

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