Understanding ECONNRESET ERROR in fluentd logs

Can someone make sense of this error?
I see this error in my fluent error only at high TPS. Not really sure which component is throwing this.

2021-08-25 15:50:01 -0700 [error]: #0 suppressed same stacktrace
2021-08-25 15:50:01 -0700 [error]: #0 unexpected error on reading data host="127.0.0.1" port=35958 error_class=Errno::ECONNRESET error="Connection reset by peer"
2021-08-25 15:50:01 -0700 [warn]: #0 emit transaction failed: error_class=Errno::ECONNRESET error="Connection reset by peer" location="<internal:prelude>:76:in `__read_nonblock'" tag="darwin.trace"
2021-08-25 15:41:39 -0700 [error]: #0 suppressed same stacktrace
2021-08-25 15:41:39 -0700 [error]: #0 unexpected error on reading data host="127.0.0.1" port=58526 error_class=Errno::ECONNRESET error="Connection reset by peer"
2021-08-25 15:41:39 -0700 [warn]: #0 emit transaction failed: error_class=Errno::ECONNRESET error="Connection reset by peer" location="<internal:prelude>:76:in `__read_nonblock'" tag="darwin.trace"
2021-08-25 15:40:25 -0700 [error]: #0 suppressed same stacktrace
2021-08-25 15:40:25 -0700 [error]: #0 unexpected error on reading data host="127.0.0.1" port=59044 error_class=Errno::ECONNRESET error="Connection reset by peer"
2021-08-25 15:40:25 -0700 [warn]: #0 emit transaction failed: error_class=Errno::ECONNRESET error="Connection reset by peer" location="<internal:prelude>:76:in `__read_nonblock'" tag="darwin.trace"2021-08-25 15:50:01 -0700 [error]: #0 suppressed same stacktrace
2021-08-25 15:50:01 -0700 [error]: #0 unexpected error on reading data host="127.0.0.1" port=35958 error_class=Errno::ECONNRESET error="Connection reset by peer"
2021-08-25 15:50:01 -0700 [warn]: #0 emit transaction failed: error_class=Errno::ECONNRESET error="Connection reset by peer" location="<internal:prelude>:76:in `__read_nonblock'" tag="darwin.trace"
2021-08-25 15:41:39 -0700 [error]: #0 suppressed same stacktrace
2021-08-25 15:41:39 -0700 [error]: #0 unexpected error on reading data host="127.0.0.1" port=58526 error_class=Errno::ECONNRESET error="Connection reset by peer"
2021-08-25 15:41:39 -0700 [warn]: #0 emit transaction failed: error_class=Errno::ECONNRESET error="Connection reset by peer" location="<internal:prelude>:76:in `__read_nonblock'" tag="darwin.trace"
2021-08-25 15:40:25 -0700 [error]: #0 suppressed same stacktrace
2021-08-25 15:40:25 -0700 [error]: #0 unexpected error on reading data host="127.0.0.1" port=59044 error_class=Errno::ECONNRESET error="Connection reset by peer"
2021-08-25 15:40:25 -0700 [warn]: #0 emit transaction failed: error_class=Errno::ECONNRESET error="Connection reset by peer" location="<internal:prelude>:76:in `__read_nonblock'" tag="darwin.trace"

Here is my config:

<source>
  @type forward
  bind 0.0.0.0
  port 24224
</source>

<source>
  @type prometheus_monitor
  interval 10
  <labels>
    hostname "#{Socket.gethostname}"
    container_name "#{ENV['CONTAINER_NAME']}"
  </labels>
</source>

<filter darwin.trace>
  @type record_transformer
  enable_ruby
  <record>
    event_timestamp ${time.strftime('%Y-%m-%d %H:%M:%S.%L %z')}
    unix_timestamp ${(Time.at(time).to_f*1000).truncate()} 
  </record>
</filter>

<filter darwin.trace>
  @type stdout
</filter>

<filter darwin.trace>
  @type prometheus
  <metric>
    name fluentd_input_status_num_records_total
    type counter
    desc The total number of incoming records
    <labels>
      tag darwin.trace
      hostname "#{Socket.gethostname}"
      container_name "#{ENV['CONTAINER_NAME']}"
    </labels>
  </metric>
</filter>

<match darwin.trace>
  @type copy
  <store>
    @type http
    @id out_http
    endpoint_url "#{ENV['LOG_PUBLISH_ENDPOINT_V2']}"
    custom_headers "#{ENV['LOGGING_SVC_HEADER_JSONSTR']}"
    serializer json
  </store>
  <store>
    @type prometheus
    <metric>
      name fluentd_output_status_num_records_total
      type counter
      desc The total number of outgoing records
      <labels>
        tag dariwn.trace
        hostname "#{Socket.gethostname}" 
        container_name "#{ENV['CONTAINER_NAME']}"
      </labels>
    </metric>
  </store>
</match>

<source>
  @type prometheus
  bind 0.0.0.0
  port 24231
  metrics_path /fluentd/metrics
  aggregated_metrics_path /fluentd/aggregated_metrics
</source>

<source>
  @type prometheus_output_monitor
  interval 10
  <labels>
    hostname "#{Socket.gethostname}"
    container_name "#{ENV['CONTAINER_NAME']}"
  </labels>
</source>