Kafka trigger: read kafka producer's error channel

[csimplestring]

Is your feature request related to a problem? Please describe.
When the kafka cluster brokers are restarted, I notice that the kafka-sensor fails to produce the messages. And no kafka-related error messages printed in log. I have to manually restart the sensor pod, then it works again.

argo-event version: 1.3.1

Describe the solution you'd like

According to the comment of sarama's AsyncProducer:

// Errors is the error output channel back to the user. You MUST read from this
// channel or the Producer will deadlock when the channel is full. Alternatively,
// you can set Producer.Return.Errors in your config to false, which prevents
// errors to be returned.

However, there is no such option to specify the Producer.Return.Errors to be False in kafka-trigger yaml template.

After checking the source code of kafka trigger , I found that the producer's error channel is not read.

Therefore I believe the most straightforward way to fix is by reading error channel, for example

// inside NewKafkaTrigger function, add below code

go func() {
	
	for err := range producer.Errors() {
		log.Println(err)
	}
}()

Describe alternatives you've considered

Additional context
The root cause is potentially related to
IBM/sarama#2133
IBM/sarama#1430
Ideally kafka async producer should refresh metadata and re-connect to brokers automatically, but seems that is not completely fixed by Shopify.

---

Message from the maintainers:

If you wish to see this enhancement implemented please add a 👍 reaction to this issue! We often sort issues this way to know what to prioritize.