Merge branch 'bugfix/produce-ack-0'

everesio · everesio · commit 0e6ee1d43129 · 2020-07-15T00:25:54.000+02:00
diff --git a/README.md b/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
diff --git a/cmd/kafka-proxy/server.go b/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")
diff --git a/config/config.go b/config/config.go
@@ -141,6 +141,9 @@ type Config struct {
 				Timeout    time.Duration
 			}
 		}
+		Producer struct {
+			Acks0Disabled bool
+		}
 	}
 	ForwardProxy struct {
 		Url string
diff --git a/proxy/client.go b/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,
diff --git a/proxy/processor.go b/proxy/processor.go
@@ -16,6 +16,7 @@ const (
 	defaultReadTimeout        = 30 * time.Second
 	minOpenRequests           = 16
 
+	apiKeyProduce        = int16(0)
 	apiKeySaslHandshake  = int16(17)
 	apiKeyApiApiVersions = int16(18)
 
@@ -40,6 +41,7 @@ type ProcessorConfig struct {
 	LocalSasl             *LocalSasl
 	AuthServer            *AuthServer
 	ForbiddenApiKeys      map[int16]struct{}
+	ProducerAcks0Disabled bool
 }
 
 type processor struct {
@@ -59,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 {
@@ -102,6 +106,7 @@ func newProcessor(cfg ProcessorConfig, brokerAddress string) *processor {
 		localSasl:                  cfg.LocalSasl,
 		authServer:                 cfg.AuthServer,
 		forbiddenApiKeys:           cfg.ForbiddenApiKeys,
+		producerAcks0Disabled:      cfg.ProducerAcks0Disabled,
 	}
 }
 
@@ -124,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)
@@ -141,6 +147,8 @@ type RequestsLoopContext struct {
 
 	localSasl     *LocalSasl
 	localSaslDone bool
+
+	producerAcks0Disabled bool
 }
 
 // used by local authentication
diff --git a/proxy/processor_default.go b/proxy/processor_default.go
@@ -1,6 +1,7 @@
 package proxy
 
 import (
+	"bytes"
 	"errors"
 	"fmt"
 	"github.com/grepplabs/kafka-proxy/proxy/protocol"
@@ -20,8 +21,12 @@ func (handler *DefaultRequestHandler) handleRequest(dst DeadlineWriter, src Dead
 	// logrus.Println("Await Kafka request")
 
 	// waiting for first bytes or EOF - reset deadlines
-	src.SetReadDeadline(time.Time{})
-	dst.SetWriteDeadline(time.Time{})
+	if err = src.SetReadDeadline(time.Time{}); err != nil {
+		return true, err
+	}
+	if err = dst.SetWriteDeadline(time.Time{}); err != nil {
+		return true, err
+	}
 
 	keyVersionBuf := make([]byte, 8) // Size => int32 + ApiKey => int16 + ApiVersion => int16
 
@@ -67,8 +72,9 @@ func (handler *DefaultRequestHandler) handleRequest(dst DeadlineWriter, src Dead
 					return true, fmt.Errorf("only saslHandshake version 0 and 1 are supported, got version %d", requestKeyVersion.ApiVersion)
 				}
 				ctx.localSaslDone = true
-				src.SetDeadline(time.Time{})
-
+				if err = src.SetDeadline(time.Time{}); err != nil {
+					return false, err
+				}
 				// defaultRequestHandler was consumed but due to local handling enqueued defaultResponseHandler will not be.
 				return false, ctx.putNextRequestHandler(defaultRequestHandler)
 			case apiKeyApiApiVersions:
@@ -79,11 +85,18 @@ func (handler *DefaultRequestHandler) handleRequest(dst DeadlineWriter, src Dead
 		}
 	}
 
-	// send inFlightRequest to channel before myCopyN to prevent race condition in proxyResponses
-	if err = sendRequestKeyVersion(ctx.openRequestsChannel, openRequestSendTimeout, requestKeyVersion); err != nil {
+	mustReply, readBytes, err := handler.mustReply(requestKeyVersion, src, ctx)
+	if err != nil {
 		return true, err
 	}
 
+	// send inFlightRequest to channel before myCopyN to prevent race condition in proxyResponses
+	if mustReply {
+		if err = sendRequestKeyVersion(ctx.openRequestsChannel, openRequestSendTimeout, requestKeyVersion); err != nil {
+			return true, err
+		}
+	}
+
 	requestDeadline := time.Now().Add(ctx.timeout)
 	err = dst.SetWriteDeadline(requestDeadline)
 	if err != nil {
@@ -98,24 +111,82 @@ func (handler *DefaultRequestHandler) handleRequest(dst DeadlineWriter, src Dead
 	if _, err = dst.Write(keyVersionBuf); err != nil {
 		return false, err
 	}
+	// write - send to broker
+	if len(readBytes) > 0 {
+		if _, err = dst.Write(readBytes); err != nil {
+			return false, err
+		}
+	}
 	// 4 bytes were written as keyVersionBuf (ApiKey, ApiVersion)
-	if readErr, err = myCopyN(dst, src, int64(requestKeyVersion.Length-4), ctx.buf); err != nil {
+	if readErr, err = myCopyN(dst, src, int64(requestKeyVersion.Length-int32(4+len(readBytes))), ctx.buf); err != nil {
 		return readErr, err
 	}
 	if requestKeyVersion.ApiKey == apiKeySaslHandshake {
 		if requestKeyVersion.ApiVersion == 0 {
 			return false, ctx.putNextHandlers(saslAuthV0RequestHandler, saslAuthV0ResponseHandler)
 		}
 	}
-	return false, ctx.putNextHandlers(defaultRequestHandler, defaultResponseHandler)
+	if mustReply {
+		return false, ctx.putNextHandlers(defaultRequestHandler, defaultResponseHandler)
+	} else {
+		return false, ctx.putNextRequestHandler(defaultRequestHandler)
+	}
+}
+
+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{}
+
+		var (
+			acks int16
+			err  error
+		)
+		var bufferRead bytes.Buffer
+		reader := io.TeeReader(src, &bufferRead)
+		switch requestKeyVersion.ApiVersion {
+		case 0, 1, 2:
+			// CorrelationID + ClientID
+			if err = acksReader.ReadAndDiscardHeaderV1Part(reader); err != nil {
+				return false, nil, err
+			}
+			// acks (INT16)
+			acks, err = acksReader.ReadAndDiscardProduceAcks(reader)
+			if err != nil {
+				return false, nil, err
+			}
+
+		case 3, 4, 5, 6, 7, 8:
+			// CorrelationID + ClientID
+			if err = acksReader.ReadAndDiscardHeaderV1Part(reader); err != nil {
+				return false, nil, err
+			}
+			// transactional_id (NULLABLE_STRING),acks (INT16)
+			acks, err = acksReader.ReadAndDiscardProduceTxnAcks(reader)
+			if err != nil {
+				return false, nil, err
+			}
+		default:
+			return false, nil, fmt.Errorf("produce version %d is not supported", requestKeyVersion.ApiVersion)
+		}
+		return acks != 0, bufferRead.Bytes(), nil
+	}
+	return true, nil, nil
 }
 
 func (handler *DefaultResponseHandler) handleResponse(dst DeadlineWriter, src DeadlineReader, ctx *ResponsesLoopContext) (readErr bool, err error) {
 	//logrus.Println("Await Kafka response")
 
 	// waiting for first bytes or EOF - reset deadlines
-	src.SetReadDeadline(time.Time{})
-	dst.SetWriteDeadline(time.Time{})
+	if err = src.SetReadDeadline(time.Time{}); err != nil {
+		return true, err
+	}
+	if err = dst.SetWriteDeadline(time.Time{}); err != nil {
+		return true, err
+	}
 
 	responseHeaderBuf := make([]byte, 8) // Size => int32, CorrelationId => int32
 	if _, err = io.ReadFull(src, responseHeaderBuf); err != nil {
diff --git a/proxy/processor_default_test.go b/proxy/processor_default_test.go
diff --git a/proxy/protocol/request_produce_reader.go b/proxy/protocol/request_produce_reader.go

Original file line number	Diff line number	Diff line change
`@@ -141,6 +141,9 @@ type Config struct {`
`141`	`141`	`Timeout time.Duration`
`142`	`142`	`}`
`143`	`143`	`}`
	`144`	`+ Producer struct {`
	`145`	`+ Acks0Disabled bool`
	`146`	`+ }`
`144`	`147`	`}`
`145`	`148`	`ForwardProxy struct {`
`146`	`149`	`Url string`