enforce idempotency config rules: acks=all, retries > 0
We do not enforce max inflight requests per connection because our
client internally bounds that to 1 unless we notice we are on Kafka
1.0.0, where the limit was raised to 5.
@@ -121,6 +121,14 @@ func (cfg *cfg) validate() error {
if cfg.disableIdempotency && cfg.txnID != nil {
return errors.New("cannot both disable idempotent writes and use transactional IDs")
}
+ if !cfg.disableIdempotency {
+ if cfg.acks.val != -1 {
+ return errors.New("idempotency requires acks=all")
+ }
+ if cfg.retries == 0 {
+ return errors.New("idempotency requires RequestRetries to be greater than 0")
+ }
+ }
for _, limit := range []struct {
name string
