Allow disabling produce required acks check

Author: everesio

README.md

@@ -109,6 +109,7 @@ See:
           --log-level-fieldname string                     Log level fieldname for json format (default "@level")
           --log-msg-fieldname string                       Message fieldname for json format (default "@message")
           --log-time-fieldname string                      Time fieldname for json format (default "@timestamp")
+          --producer-acks-0-disabled                       Assume fire-and-forget is never sent by the producer. Enabling this parameter will increase performance
           --proxy-listener-ca-chain-cert-file string       PEM encoded CA's certificate file. If provided, client certificate is required and verified
           --proxy-listener-cert-file string                PEM encoded file with server certificate
           --proxy-listener-cipher-suites stringSlice       List of supported cipher suites

cmd/kafka-proxy/server.go

@@ -142,6 +142,8 @@ func initFlags() {
 	// http://kafka.apache.org/protocol.html#protocol_api_keys
 	Server.Flags().IntSliceVar(&c.Kafka.ForbiddenApiKeys, "forbidden-api-keys", []int{}, "Forbidden Kafka request types. The restriction should prevent some Kafka operations e.g. 20 - DeleteTopics")
 
+	Server.Flags().BoolVar(&c.Kafka.Producer.Acks0Disabled, "producer-acks-0-disabled", false, "Assume fire-and-forget is never sent by the producer. Enabling this parameter will increase performance")
+
 	// TLS
 	Server.Flags().BoolVar(&c.Kafka.TLS.Enable, "tls-enable", false, "Whether or not to use TLS when connecting to the broker")
 	Server.Flags().BoolVar(&c.Kafka.TLS.InsecureSkipVerify, "tls-insecure-skip-verify", false, "It controls whether a client verifies the server's certificate chain and host name")

config/config.go

@@ -141,6 +141,9 @@ type Config struct {
 				Timeout    time.Duration
 			}
 		}
+		Producer struct {
+			Acks0Disabled bool
+		}
 	}
 	ForwardProxy struct {
 		Url string

proxy/client.go

@@ -155,7 +155,8 @@ func NewClient(conns *ConnSet, c *config.Config, netAddressMappingFunc config.Ne
 				timeout:   c.Auth.Gateway.Server.Timeout,
 				tokenInfo: gatewayTokenInfo,
 			},
-			ForbiddenApiKeys: forbiddenApiKeys,
+			ForbiddenApiKeys:      forbiddenApiKeys,
+			ProducerAcks0Disabled: c.Kafka.Producer.Acks0Disabled,
 		},
 		dialAddressMapping: dialAddressMapping,
 		kafkaClientCert:    kafkaClientCert,

proxy/processor.go

@@ -41,6 +41,7 @@ type ProcessorConfig struct {
 	LocalSasl             *LocalSasl
 	AuthServer            *AuthServer
 	ForbiddenApiKeys      map[int16]struct{}
+	ProducerAcks0Disabled bool
 }
 
 type processor struct {
@@ -60,6 +61,8 @@ type processor struct {
 	forbiddenApiKeys map[int16]struct{}
 	// metrics
 	brokerAddress string
+	// producer will never send request with acks=0
+	producerAcks0Disabled bool
 }
 
 func newProcessor(cfg ProcessorConfig, brokerAddress string) *processor {
@@ -103,6 +106,7 @@ func newProcessor(cfg ProcessorConfig, brokerAddress string) *processor {
 		localSasl:                  cfg.LocalSasl,
 		authServer:                 cfg.AuthServer,
 		forbiddenApiKeys:           cfg.ForbiddenApiKeys,
+		producerAcks0Disabled:      cfg.ProducerAcks0Disabled,
 	}
 }
 
@@ -125,6 +129,7 @@ func (p *processor) RequestsLoop(dst DeadlineWriter, src DeadlineReaderWriter) (
 		buf:                        make([]byte, p.requestBufferSize),
 		localSasl:                  p.localSasl,
 		localSaslDone:              false, // sequential processing - mutex is required
+		producerAcks0Disabled:      p.producerAcks0Disabled,
 	}
 
 	return ctx.requestsLoop(dst, src)
@@ -142,6 +147,8 @@ type RequestsLoopContext struct {
 
 	localSasl     *LocalSasl
 	localSaslDone bool
+
+	producerAcks0Disabled bool
 }
 
 // used by local authentication

proxy/processor_default.go

@@ -85,7 +85,7 @@ func (handler *DefaultRequestHandler) handleRequest(dst DeadlineWriter, src Dead
 		}
 	}
 
-	mustReply, readBytes, err := handler.mustReply(requestKeyVersion, src)
+	mustReply, readBytes, err := handler.mustReply(requestKeyVersion, src, ctx)
 	if err != nil {
 		return true, err
 	}
@@ -133,8 +133,11 @@ func (handler *DefaultRequestHandler) handleRequest(dst DeadlineWriter, src Dead
 	}
 }
 
-func (handler *DefaultRequestHandler) mustReply(requestKeyVersion *protocol.RequestKeyVersion, src io.Reader) (bool, []byte, error) {
+func (handler *DefaultRequestHandler) mustReply(requestKeyVersion *protocol.RequestKeyVersion, src io.Reader, ctx *RequestsLoopContext) (bool, []byte, error) {
 	if requestKeyVersion.ApiKey == apiKeyProduce {
+		if ctx.producerAcks0Disabled {
+			return true, nil, nil
+		}
 		// header version for produce [0..8] is 1 (request_api_key,request_api_version,correlation_id (INT32),client_id, NULLABLE_STRING )
 		acksReader := protocol.RequestAcksReader{}