Merge pull request #178 from grepplabs/kafka-3.8

kafka 3.8 / 3.9 support

Dockerfile

@@ -1,4 +1,4 @@
-FROM --platform=$BUILDPLATFORM golang:1.22-alpine3.19 as builder
+FROM --platform=$BUILDPLATFORM golang:1.22-alpine3.19 AS builder
 RUN apk add alpine-sdk ca-certificates
 
 ARG TARGETOS

Dockerfile.all

@@ -1,4 +1,4 @@
-FROM --platform=$BUILDPLATFORM golang:1.22-alpine3.19 as builder
+FROM --platform=$BUILDPLATFORM golang:1.22-alpine3.19 AS builder
 RUN apk add alpine-sdk ca-certificates
 
 ARG TARGETOS

Makefile

@@ -10,7 +10,7 @@ VERSION       ?= $(shell git describe --tags --always --dirty)
 GOPKGS         = $(shell go list ./... | grep -v /vendor/)
 BUILD_FLAGS   ?=
 LDFLAGS       ?= -X github.com/grepplabs/kafka-proxy/config.Version=$(VERSION) -w -s
-TAG           ?= "v0.3.11"
+TAG           ?= "v0.3.12"
 GOOS          ?= $(if $(TARGETOS),$(TARGETOS),linux)
 GOARCH        ?= $(if $(TARGETARCH),$(TARGETARCH),amd64)
 GOARM         ?= $(TARGETVARIANT)

README.md

@@ -40,18 +40,19 @@ As not every Kafka release adds new messages/versions which are relevant to the
 | 0.2.9               | to 2.8.0      |
 | 0.3.1               | to 3.4.0      |
 | 0.3.11              | to 3.7.0      |
+| 0.3.12              | to 3.9.0      |
 
 ### Install binary release
 
 1. Download the latest release
 
    Linux
 
-        curl -Ls https://github.com/grepplabs/kafka-proxy/releases/download/v0.3.11/kafka-proxy-v0.3.11-linux-amd64.tar.gz | tar xz
+        curl -Ls https://github.com/grepplabs/kafka-proxy/releases/download/v0.3.12/kafka-proxy-v0.3.12-linux-amd64.tar.gz | tar xz
 
    macOS
 
-        curl -Ls https://github.com/grepplabs/kafka-proxy/releases/download/v0.3.11/kafka-proxy-v0.3.11-darwin-amd64.tar.gz | tar xz
+        curl -Ls https://github.com/grepplabs/kafka-proxy/releases/download/v0.3.12/kafka-proxy-v0.3.12-darwin-amd64.tar.gz | tar xz
 
 2. Move the binary in to your PATH.
 
@@ -69,7 +70,7 @@ Docker images are available on [Docker Hub](https://hub.docker.com/r/grepplabs/k
 
 You can launch a kafka-proxy container for trying it out with
 
-    docker run --rm -p 30001-30003:30001-30003 grepplabs/kafka-proxy:0.3.11 \
+    docker run --rm -p 30001-30003:30001-30003 grepplabs/kafka-proxy:0.3.12 \
               server \
             --bootstrap-server-mapping "localhost:19092,0.0.0.0:30001" \
             --bootstrap-server-mapping "localhost:29092,0.0.0.0:30002" \
@@ -88,7 +89,7 @@ Docker images with precompiled plugins located in `/opt/kafka-proxy/bin/` are ta
 
 You can launch a kafka-proxy container with auth-ldap plugin for trying it out with
 
-    docker run --rm -p 30001-30003:30001-30003 grepplabs/kafka-proxy:0.3.11-all \
+    docker run --rm -p 30001-30003:30001-30003 grepplabs/kafka-proxy:0.3.12-all \
                   server \
                 --bootstrap-server-mapping "localhost:19092,0.0.0.0:30001" \
                 --bootstrap-server-mapping "localhost:29092,0.0.0.0:30002" \

cmd/plugin-auth-ldap/main.go

@@ -11,7 +11,6 @@ import (
 	"github.com/hashicorp/go-plugin"
 	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
-	"io/ioutil"
 	"net"
 	"net/url"
 	"os"
@@ -317,7 +316,7 @@ func getTlsConfig(caCertFile string, insecureSkipVerify bool) (*tls.Config, erro
 	if caCertFile == "" {
 		return &tls.Config{InsecureSkipVerify: insecureSkipVerify}, nil
 	} else {
-		certData, err := ioutil.ReadFile(caCertFile)
+		certData, err := os.ReadFile(caCertFile)
 		if err != nil {
 			return nil, errors.Wrapf(err, "reading certificate file %s", caCertFile)
 		}

config/jaas.go

@@ -3,7 +3,7 @@ package config
 import (
 	"errors"
 	"fmt"
-	"io/ioutil"
+	"os"
 	"regexp"
 	"strings"
 )
@@ -19,7 +19,7 @@ type JaasCredentials struct {
 }
 
 func NewJaasCredentialFromFile(filename string) (*JaasCredentials, error) {
-	bytes, err := ioutil.ReadFile(filename)
+	bytes, err := os.ReadFile(filename)
 	if err != nil {
 		return nil, err
 	}

config/jaas_test.go

@@ -1,10 +1,10 @@
 package config
 
 import (
-	"github.com/stretchr/testify/assert"
-	"io/ioutil"
 	"os"
 	"testing"
+
+	"github.com/stretchr/testify/assert"
 )
 
 func TestExtractsJaasCredentials(t *testing.T) {
@@ -30,7 +30,7 @@ func TestExtractsJaasCredentialsFromFile(t *testing.T) {
 		  password="veyaiThai5que0ieb5le";
 		};
 	`
-	tmpFile, err := ioutil.TempFile("", "kafka-proxy-jaas-test")
+	tmpFile, err := os.CreateTemp("", "kafka-proxy-jaas-test")
 	assert.Nil(t, err)
 	defer os.Remove(tmpFile.Name())
 	defer tmpFile.Close()

pkg/libs/googleid/certs.go

@@ -7,7 +7,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"golang.org/x/net/context/ctxhttp"
-	"io/ioutil"
+	"io"
 	"math/big"
 	"net/http"
 	"time"
@@ -59,7 +59,7 @@ func GetCerts(ctx context.Context) (*Certs, error) {
 	}
 	defer resp.Body.Close()
 
-	body, err := ioutil.ReadAll(resp.Body)
+	body, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return nil, err
 	}

pkg/libs/googleid/service_account.go

@@ -12,9 +12,10 @@ import (
 	"golang.org/x/oauth2/google"
 	"golang.org/x/oauth2/jws"
 	"google.golang.org/api/oauth2/v2"
-	"io/ioutil"
+	"io"
 	"net/http"
 	"net/url"
+	"os"
 	"strings"
 	"time"
 )
@@ -32,7 +33,7 @@ type ServiceAccountTokenSource struct {
 }
 
 func NewServiceAccountTokenSource(credentialsFile string, targetAudience string) (*ServiceAccountTokenSource, error) {
-	data, err := ioutil.ReadFile(credentialsFile)
+	data, err := os.ReadFile(credentialsFile)
 	if err != nil {
 		return nil, err
 	}
@@ -130,7 +131,7 @@ func doExchange(ctx context.Context, token string) ([]byte, error) {
 	}
 	defer resp.Body.Close()
 
-	body, err := ioutil.ReadAll(resp.Body)
+	body, err := io.ReadAll(resp.Body)
 	if err != nil {
 		return nil, err
 	}

pkg/libs/oidc-provider/plugin.go

@@ -4,7 +4,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io/ioutil"
 	"os"
 	"sync"
 	"time"
@@ -210,7 +209,7 @@ func getTokenResponse(token string, status int) (apis.TokenResponse, error) {
 }
 
 func getTokenSource(credentialsFilePath string, targetAud string) (idTokenSource, error) {
-	data, err := ioutil.ReadFile(credentialsFilePath)
+	data, err := os.ReadFile(credentialsFilePath)
 
 	if err != nil {
 		return nil, err

pkg/libs/oidc/password_grant.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"io/ioutil"
+	"os"
 
 	"github.com/sirupsen/logrus"
 	"golang.org/x/oauth2"
@@ -28,7 +28,7 @@ type PasswordGrantTokenSource struct {
 }
 
 func NewPasswordGrantTokenSource(credentialsFile string, targetAudience string) (*PasswordGrantTokenSource, error) {
-	data, err := ioutil.ReadFile(credentialsFile)
+	data, err := os.ReadFile(credentialsFile)
 	source := &PasswordGrantTokenSource{}
 
 	if err != nil {

pkg/libs/oidc/service_account.go

@@ -4,7 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
-	"io/ioutil"
+	"os"
 
 	"github.com/sirupsen/logrus"
 	"golang.org/x/oauth2/clientcredentials"
@@ -26,7 +26,7 @@ type ServiceAccountTokenSource struct {
 }
 
 func NewServiceAccountTokenSource(credentialsFile string, targetAudience string) (*ServiceAccountTokenSource, error) {
-	data, err := ioutil.ReadFile(credentialsFile)
+	data, err := os.ReadFile(credentialsFile)
 	source := &ServiceAccountTokenSource{}
 
 	if err != nil {

pkg/libs/util/watcher_test.go

@@ -3,7 +3,6 @@ package util
 import (
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
-	"io/ioutil"
 	"os"
 	"path/filepath"
 	"sync/atomic"
@@ -16,18 +15,18 @@ func TestWatchRegularFileChange(t *testing.T) {
 
 	a := assert.New(t)
 
-	dirName, err := ioutil.TempDir("", "watcher-test-")
+	dirName, err := os.MkdirTemp("", "watcher-test-")
 	a.Nil(err)
 	defer os.Remove(dirName)
 
-	targetSecret, err := ioutil.TempFile(dirName, "secret-")
+	targetSecret, err := os.CreateTemp(dirName, "secret-")
 	a.Nil(err)
 	defer os.Remove(targetSecret.Name())
 
 	_, err = targetSecret.WriteString("secret1")
 	a.Nil(err)
 
-	data, err := ioutil.ReadFile(targetSecret.Name())
+	data, err := os.ReadFile(targetSecret.Name())
 	a.Nil(err)
 	a.Equal("secret1", string(data))
 
@@ -59,7 +58,7 @@ func TestWatchRegularFileChange(t *testing.T) {
 	opsFinal := atomic.LoadInt32(&ops)
 	a.Equal(int32(1), opsFinal)
 
-	data, err = ioutil.ReadFile(targetSecret.Name())
+	data, err = os.ReadFile(targetSecret.Name())
 	a.Nil(err)
 	a.Equal("secret1addition", string(data))
 }
@@ -74,23 +73,23 @@ func TestWatchLinkedFileChange(t *testing.T) {
 	*/
 	a := assert.New(t)
 
-	dirName, err := ioutil.TempDir("", "watcher-test-")
+	dirName, err := os.MkdirTemp("", "watcher-test-")
 	a.Nil(err)
 	defer os.Remove(dirName)
 
-	dirTmp1, err := ioutil.TempDir(dirName, "tmp1-")
+	dirTmp1, err := os.MkdirTemp(dirName, "tmp1-")
 	a.Nil(err)
 	defer os.Remove(dirTmp1)
 
-	dirTmp2, err := ioutil.TempDir(dirName, "tmp2-")
+	dirTmp2, err := os.MkdirTemp(dirName, "tmp2-")
 	a.Nil(err)
 	defer os.Remove(dirTmp2)
 
-	targetSecret1, err := ioutil.TempFile(dirTmp1, "secret-")
+	targetSecret1, err := os.CreateTemp(dirTmp1, "secret-")
 	a.Nil(err)
 	defer os.Remove(targetSecret1.Name())
 
-	targetSecret2, err := ioutil.TempFile(dirTmp2, "secret-")
+	targetSecret2, err := os.CreateTemp(dirTmp2, "secret-")
 	a.Nil(err)
 	defer os.Remove(targetSecret2.Name())
 
@@ -110,7 +109,7 @@ func TestWatchLinkedFileChange(t *testing.T) {
 	a.Nil(err)
 	defer os.Remove(secretLink)
 
-	data, err := ioutil.ReadFile(secretLink)
+	data, err := os.ReadFile(secretLink)
 	a.Nil(err)
 	a.Equal("secret1", string(data))
 
@@ -145,7 +144,7 @@ func TestWatchLinkedFileChange(t *testing.T) {
 	opsFinal := atomic.LoadInt32(&ops)
 	a.Equal(int32(1), opsFinal)
 
-	data, err = ioutil.ReadFile(secretLink)
+	data, err = os.ReadFile(secretLink)
 	a.Nil(err)
 	a.Equal("secret2", string(data))
 }

proxy/processor_default.go

@@ -159,7 +159,7 @@ func (handler *DefaultRequestHandler) mustReply(requestKeyVersion *protocol.Requ
 				return false, nil, err
 			}
 
-		case 3, 4, 5, 6, 7, 8, 9, 10:
+		case 3, 4, 5, 6, 7, 8, 9, 10, 11:
 			// CorrelationID + ClientID
 			if err = acksReader.ReadAndDiscardHeaderV1Part(reader); err != nil {
 				return false, nil, err

proxy/protocol/request_key_version.go

@@ -360,6 +360,32 @@ func (r *RequestKeyVersion) ResponseHeaderVersion() int16 {
 		return 1
 	case 74: // ListClientMetricsResources
 		return 1
+	case 75: // DescribeTopicPartitions
+		return 1
+	case 76: // ShareGroupHeartbeat
+		return 1
+	case 77: // ShareGroupDescribe
+		return 1
+	case 78: // ShareFetch
+		return 1
+	case 79: // ShareAcknowledge
+		return 1
+	case 80: // AddRaftVoter
+		return 1
+	case 81: // RemoveRaftVoter
+		return 1
+	case 82: // UpdateRaftVoter
+		return 1
+	case 83: // InitializeShareGroupState
+		return 1
+	case 84: // ReadShareGroupState
+		return 1
+	case 85: // WriteShareGroupState
+		return 1
+	case 86: // DeleteShareGroupState
+		return 1
+	case 87: // ReadShareGroupStateSummary
+		return 1
 	default:
 		// throw new UnsupportedVersionException("Unsupported API key " + apiKey);
 		return -1

proxy/protocol/request_produce_reader.go

@@ -3,7 +3,6 @@ package protocol
 import (
 	"encoding/binary"
 	"io"
-	"io/ioutil"
 )
 
 type RequestAcksReader struct {
@@ -18,7 +17,7 @@ func (r RequestAcksReader) readAndDiscardNullableString(reader io.Reader) (err e
 		return errInvalidStringLength
 	}
 	if length > 0 {
-		if _, err = io.CopyN(ioutil.Discard, reader, int64(length)); err != nil {
+		if _, err = io.CopyN(io.Discard, reader, int64(length)); err != nil {
 			return err
 		}
 	}

proxy/protocol/response_header.go

@@ -6,7 +6,6 @@ import (
 	"fmt"
 	"github.com/pkg/errors"
 	"io"
-	"io/ioutil"
 )
 
 type ResponseHeader struct {
@@ -78,7 +77,7 @@ func (r *ResponseHeaderTaggedFields) MaybeRead(reader io.Reader) ([]byte, error)
 			} else if size == 0 {
 				continue
 			} else {
-				if _, err := io.CopyN(ioutil.Discard, reader, int64(size)); err != nil {
+				if _, err := io.CopyN(io.Discard, reader, int64(size)); err != nil {
 					return nil, errors.Wrap(err, "error while reading tagged field data")
 				}
 			}

proxy/protocol/responses.go

@@ -293,7 +293,10 @@ func createFindCoordinatorResponseSchemaVersions() []Schema {
 		&CompactArray{Name: coordinatorsKeyName, Ty: findCoordinatorCoordinatorsSchema4},
 		&SchemaTaggedFields{Name: "response_tagged_fields"},
 	)
-	return []Schema{findCoordinatorResponseV0, findCoordinatorResponseV1, findCoordinatorResponseV2, findCoordinatorResponseV3, findCoordinatorResponseV4}
+	findCoordinatorResponseV5 := findCoordinatorResponseV4
+	findCoordinatorResponseV6 := findCoordinatorResponseV5
+
+	return []Schema{findCoordinatorResponseV0, findCoordinatorResponseV1, findCoordinatorResponseV2, findCoordinatorResponseV3, findCoordinatorResponseV4, findCoordinatorResponseV5, findCoordinatorResponseV6}
 }
 
 func modifyMetadataResponse(decodedStruct *Struct, fn config.NetAddressMappingFunc) error {