Dataproc任务从本地Kafka同步数据至BigQuery时出现连接异常失败

Hey Bruno, let's dig into those ConnectionRefused and Connection reset by peer errors you're seeing—these are classic network/configuration hiccups, and we can work through them step by step.

First: Fixing the ConnectionRefused Error #

Your specific error mentions worker-3 trying to connect to worker-1 on port 50267 and getting rejected. Here's what to check first:

• Verify Kafka Broker Listener Configuration
  That port 50267 looks like a non-standard Kafka port—if your Kafka broker is running on worker-1, double-check its server.properties for listeners and advertised.listeners. If the broker is only bound to localhost or worker-1's loopback IP, other workers in the Dataproc cluster won't be able to reach it. Make sure advertised.listeners uses a hostname/IP that's resolvable across all cluster nodes (like the internal Dataproc hostname or worker-1's internal IP).

• Test Cluster Network Connectivity
  Log into worker-3 and run a quick connectivity test to rule out firewall/security group issues:

  nc -zv <worker-1-ip> 50267
  

  If this fails, check:

  • Dataproc cluster security group rules: Ensure worker nodes allow inbound traffic on port 50267 from other cluster nodes.
  • Worker-1's local firewall: Tools like iptables might be blocking incoming connections to that port. Run iptables -L on worker-1 to verify.

• Check if Kafka is Actually Listening on That Port
  On worker-1, confirm the Kafka process is active and listening on 50267:

  ss -tulpn | grep 50267
  

  If no process shows up, your Kafka broker might have crashed or failed to start properly. Check the Kafka logs (usually in /var/log/kafka/ by default) to find out why it didn't bind to the port.

Next: Resolving Connection reset by peer #

This error happens when an established connection gets abruptly closed by one end. Common causes and fixes:

• Kafka Broker Idle Timeouts or Resource Limits
  Kafka's connections.max.idle.ms setting might be too low, causing it to drop idle connections automatically. Check your server.properties and increase this value if needed. Also, verify the broker has enough system resources:

  • Disk space: Run df -h to ensure Kafka's log directory isn't full.
  • File handles: Use ulimit -n to check if the open file limit is high enough (Kafka needs many handles for connections and logs).
  • Memory: Run free -m to confirm the broker isn't swapping memory, which can cause unexpected connection drops.

• Unstable Cluster Network
  Temporary network blips between Dataproc workers can trigger this. Use tools like mtr <worker-1-ip> on worker-3 to monitor packet loss over time. You can also check your cloud provider's VPC monitoring dashboard for any network anomalies in your cluster's subnet.

• Tweak Kafka Client Configuration
  If your Dataproc job uses Spark/Flink to consume from Kafka, adjust the client settings to handle transient issues:

  • Increase request.timeout.ms and session.timeout.ms to give more time for connections to stabilize.
  • Enable retries with retries and retry.backoff.ms so the job doesn't fail immediately when a connection resets.

General Best Practices #

• Enable Debug Logging
  Crank up the Kafka client log level (set org.apache.kafka to DEBUG in your job's logging config) to get detailed connection logs—this will show exactly when and why connections are failing.
• Add Retry Logic in Your Job
  Wrap your Kafka consumer logic in a retry loop (or use your processing framework's built-in retry mechanisms) to handle temporary failures without crashing the entire job.
• Monitor Key Metrics
  Use Dataproc's built-in monitoring or tools like Prometheus + Grafana to track Kafka broker health (e.g., under-replicated partitions, active connections) and cluster resource usage.

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