Fix NOT_LEADER_FOR_PARTITION error with Confluent Cloud

[kamir]

This commit addresses the "Not Leader For Partition" error that occurs when using kshark with Confluent Cloud clusters.

Root Cause:

• The Writer's Transport was not properly using the configured Dialer
• Transport created new connections without SASL/TLS from the Dialer
• Bootstrap servers were used directly instead of allowing metadata discovery of actual partition leader brokers

Changes:

1. Configure Transport to use Dialer's DialContext method

   • Ensures all connections use proper SASL/TLS authentication
   • Enables correct broker discovery for managed Kafka services
   • Fixes metadata refresh for Confluent Cloud's dynamic brokers

2. Add retry logic for metadata-related errors

   • Retry up to 3 times with exponential backoff (500ms, 1s, 1.5s)
   • Specifically handle NOT_LEADER_FOR_PARTITION and LeaderNotAvailable
   • Non-retryable errors fail immediately without wasted retries
   • Detailed logging for debugging retry attempts

References:

• Seeing - "Not Leader For Partition: the client's metadata are likely out of date" error segmentio/kafka-go#1078
• Error writing message: kafka server: Tried to send a m... Your metadata is out of date. segmentio/kafka-go#712

https://claude.ai/code/session_01VSnKepTAD53ZUfXYDDvXwa