fix(config): relax ClientID validation after 1.0.0

The original validation regex was based on the one that existed in the
Java clients pre-1.0.0 but KIP-190 removed the client-side validation
because instead the brokers were updated to be able to cope with any
clientID and sanitize it before using it in metrics etc.

We can do similar in Sarama and only do client-side validation when the
user has specified a version number older than 1.0.0

Fixes #2697

Signed-off-by: Dominic Evans <[email protected]>

config.go

@@ -15,7 +15,9 @@ import (
 
 const defaultClientID = "sarama"
 
-var validID = regexp.MustCompile(`\A[A-Za-z0-9._-]+\z`)
+// validClientID specifies the permitted characters for a client.id when
+// connecting to Kafka versions before 1.0.0 (KIP-190)
+var validClientID = regexp.MustCompile(`\A[A-Za-z0-9._-]+\z`)
 
 // Config is used to pass multiple configuration options to Sarama's constructors.
 type Config struct {
@@ -846,8 +848,11 @@ func (c *Config) Validate() error {
 	switch {
 	case c.ChannelBufferSize < 0:
 		return ConfigurationError("ChannelBufferSize must be >= 0")
-	case !validID.MatchString(c.ClientID):
-		return ConfigurationError("ClientID is invalid")
+	}
+
+	// only validate clientID locally for Kafka versions before KIP-190 was implemented
+	if !c.Version.IsAtLeast(V1_0_0_0) && !validClientID.MatchString(c.ClientID) {
+		return ConfigurationError(fmt.Sprintf("ClientID value %q is not valid for Kafka versions before 1.0.0", c.ClientID))
 	}
 
 	return nil

config_test.go

@@ -2,11 +2,13 @@ package sarama
 
 import (
 	"errors"
+	"fmt"
 	"os"
 	"strings"
 	"testing"
 
 	"github.com/rcrowley/go-metrics"
+	assert "github.com/stretchr/testify/require"
 )
 
 // NewTestConfig returns a config meant to be used by tests.
@@ -30,23 +32,23 @@ func TestDefaultConfigValidates(t *testing.T) {
 	}
 }
 
-func TestInvalidClientIDConfigValidates(t *testing.T) {
-	config := NewTestConfig()
-	config.ClientID = "foo:bar"
-	err := config.Validate()
-	var target ConfigurationError
-	if !errors.As(err, &target) || string(target) != "ClientID is invalid" {
-		t.Error("Expected invalid ClientID, got ", err)
-	}
-}
-
-func TestEmptyClientIDConfigValidates(t *testing.T) {
-	config := NewTestConfig()
-	config.ClientID = ""
-	err := config.Validate()
-	var target ConfigurationError
-	if !errors.As(err, &target) || string(target) != "ClientID is invalid" {
-		t.Error("Expected invalid ClientID, got ", err)
+// TestInvalidClientIDValidated ensures that the ClientID field is checked
+// when Version is set to anything less than 1_0_0_0, but otherwise accepted
+func TestInvalidClientIDValidated(t *testing.T) {
+	for _, version := range SupportedVersions {
+		for _, clientID := range []string{"", "foo:bar", "foo|bar"} {
+			config := NewTestConfig()
+			config.ClientID = clientID
+			config.Version = version
+			err := config.Validate()
+			if config.Version.IsAtLeast(V1_0_0_0) {
+				assert.NoError(t, err)
+				continue
+			}
+			var target ConfigurationError
+			assert.ErrorAs(t, err, &target)
+			assert.ErrorContains(t, err, fmt.Sprintf("ClientID value %q is not valid for Kafka versions before 1.0.0", clientID))
+		}
 	}
 }
 

mocks/async_producer_test.go

@@ -251,14 +251,15 @@ func (brokePartitioner) RequiresConsistency() bool { return false }
 func TestProducerWithInvalidConfiguration(t *testing.T) {
 	trm := newTestReporterMock()
 	config := NewTestConfig()
-	config.ClientID = "not a valid client ID"
+	config.Version = sarama.V0_11_0_2
+	config.ClientID = "not a valid producer ID"
 	mp := NewAsyncProducer(trm, config)
 	if err := mp.Close(); err != nil {
 		t.Error(err)
 	}
 	if len(trm.errors) != 1 {
 		t.Error("Expected to report a single error")
-	} else if !strings.Contains(trm.errors[0], "ClientID is invalid") {
+	} else if !strings.Contains(trm.errors[0], `ClientID value "not a valid producer ID" is not valid for Kafka versions before 1.0.0`) {
 		t.Errorf("Unexpected error: %s", trm.errors[0])
 	}
 }

mocks/consumer_test.go

@@ -405,15 +405,16 @@ func TestConsumerOffsetsAreManagedCorrectlyWithSpecifiedOffset(t *testing.T) {
 func TestConsumerInvalidConfiguration(t *testing.T) {
 	trm := newTestReporterMock()
 	config := NewTestConfig()
-	config.ClientID = "not a valid client ID"
+	config.Version = sarama.V0_11_0_2
+	config.ClientID = "not a valid consumer ID"
 	consumer := NewConsumer(trm, config)
 	if err := consumer.Close(); err != nil {
 		t.Error(err)
 	}
 
 	if len(trm.errors) != 1 {
 		t.Error("Expected to report a single error")
-	} else if !strings.Contains(trm.errors[0], "ClientID is invalid") {
+	} else if !strings.Contains(trm.errors[0], `ClientID value "not a valid consumer ID" is not valid for Kafka versions before 1.0.0`) {
 		t.Errorf("Unexpected error: %s", trm.errors[0])
 	}
 }

mocks/sync_producer_test.go

@@ -356,15 +356,16 @@ func (f faultyEncoder) Length() int {
 func TestSyncProducerInvalidConfiguration(t *testing.T) {
 	trm := newTestReporterMock()
 	config := NewTestConfig()
-	config.ClientID = "not a valid client ID"
+	config.Version = sarama.V0_11_0_2
+	config.ClientID = "not a valid producer ID"
 	mp := NewSyncProducer(trm, config)
 	if err := mp.Close(); err != nil {
 		t.Error(err)
 	}
 
 	if len(trm.errors) != 1 {
 		t.Error("Expected to report a single error")
-	} else if !strings.Contains(trm.errors[0], "ClientID is invalid") {
+	} else if !strings.Contains(trm.errors[0], `ClientID value "not a valid producer ID" is not valid for Kafka versions before 1.0.0`) {
 		t.Errorf("Unexpected error: %s", trm.errors[0])
 	}
 }