Fix max open requests to one if idempotent writes is set to true (inf…

…luxdata#8954)

plugins/common/kafka/config.go

@@ -38,6 +38,9 @@ func (k *WriteConfig) SetConfig(config *sarama.Config) error {
 		config.Producer.MaxMessageBytes = k.MaxMessageBytes
 	}
 	config.Producer.RequiredAcks = sarama.RequiredAcks(k.RequiredAcks)
+	if config.Producer.Idempotent {
+		config.Net.MaxOpenRequests = 1
+	}
 	return k.Config.SetConfig(config)
 }