Repository: apache/pulsar-client-go
Pull request #778 (2022-05-25 07:01:04)

feat: add basic authentication
### Motivation

Miss the basic authentication.

### Modifications

- Add the basic authentication provider in `internal/auth`
- Add factory method of basic authentication provider in `client.go`
- Configure the basic authentication provider in integration tests

### Documentation

  - Does this pull request introduce a new feature? yes
  - If yes, how is the feature documented? docs
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation



Patch:
diff --git a/.github/workflows/go.yml b/.github/workflows/go.yml
index a3c4c60..1819b53 100644
--- a/.github/workflows/go.yml
+++ b/.github/workflows/go.yml
@@ -6,7 +6,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        go-version: [1.15, 1.16, 1.17]
+        go-version: [1.15, 1.16, 1.17, 1.18]
     steps:
       - name: clean docker cache
         run: |
diff --git a/.github/workflows/project.yml b/.github/workflows/project.yml
index 810979a..ce8545b 100644
--- a/.github/workflows/project.yml
+++ b/.github/workflows/project.yml
@@ -6,7 +6,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        go-version: [1.15, 1.16, 1.17]
+        go-version: [1.15, 1.16, 1.17, 1.18]
     steps:
       - name: Set up Go
         uses: actions/setup-go@v1
@@ -19,7 +19,7 @@ jobs:
 
       - name: InstallTool
         run: |
-          wget -O - -q https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh| sh -s v1.18.0
+          wget -O - -q https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh| sh -s v1.46.2
           ./bin/golangci-lint --version
 
       - name: Build
diff --git a/.golangci.yml b/.golangci.yml
index e87841b..525d34c 100644
--- a/.golangci.yml
+++ b/.golangci.yml
@@ -10,10 +10,9 @@ linters:
     - bodyclose
     - deadcode
     - goimports
-    - golint
+    - revive
     - gosimple
     - govet
-    - interfacer
     - misspell
     - structcheck
     - stylecheck
diff --git a/Dockerfile b/Dockerfile
index e12cc80..6dd5817 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -30,6 +30,8 @@ COPY integration-tests/certs /pulsar/certs
 COPY integration-tests/tokens /pulsar/tokens
 COPY integration-tests/standalone.conf /pulsar/conf
 COPY integration-tests/client.conf /pulsar/conf
+COPY integration-tests/.htpasswd /pulsar/conf
+ENV PULSAR_EXTRA_OPTS="-Dpulsar.auth.basic.conf=/pulsar/conf/.htpasswd"
 COPY pulsar-test-service-start.sh /pulsar/bin
 COPY pulsar-test-service-stop.sh /pulsar/bin
 COPY run-ci.sh /pulsar/bin
diff --git a/pulsar/client.go b/pulsar/client.go
index f4642c6..bc05b25 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -78,6 +78,11 @@ func NewAuthenticationOAuth2(authParams map[string]string) Authentication {
 	return oauth
 }
 
+// NewAuthenticationBasic Creates Basic Authentication provider
+func NewAuthenticationBasic(username, password string) (Authentication, error) {
+	return auth.NewAuthenticationBasic(username, password)
+}
+
 // ClientOptions is used to construct a Pulsar Client instance.
 type ClientOptions struct {
 	// Configure the service URL for the Pulsar service.
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 06ccfd5..4cc1645 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -534,6 +534,7 @@ func (pc *partitionConsumer) internalAck(req *ackRequest) {
 	}
 
 	if pc.options.ackWithResponse {
+		cmdAck.RequestId = proto.Uint64(reqID)
 		_, err := pc.client.rpcClient.RequestOnCnx(pc._getConn(), reqID, pb.BaseCommand_ACK, cmdAck)
 		if err != nil {
 			pc.log.WithError(err).Error("Ack with response error")
@@ -755,20 +756,20 @@ func createEncryptionContext(msgMeta *pb.MessageMetadata) *EncryptionContext {
 		encCtx.CompressionType = CompressionType(*msgMeta.Compression)
 	}
 
-	kMap := map[string]EncryptionKey{}
+	keyMap := map[string]EncryptionKey{}
 	for _, k := range msgMeta.GetEncryptionKeys() {
 		metaMap := map[string]string{}
 		for _, m := range k.GetMetadata() {
 			metaMap[*m.Key] = *m.Value
 		}
 
-		kMap[*k.Key] = EncryptionKey{
+		keyMap[*k.Key] = EncryptionKey{
 			KeyValue: k.GetValue(),
 			Metadata: metaMap,
 		}
 	}
 
-	encCtx.Keys = kMap
+	encCtx.Keys = keyMap
 	return &encCtx
 }
 
@@ -1186,7 +1187,11 @@ func (pc *partitionConsumer) grabConn() error {
 
 	pc._setConn(res.Cnx)
 	pc.log.Info("Connected consumer")
-	pc._getConn().AddConsumeHandler(pc.consumerID, pc)
+	err = pc._getConn().AddConsumeHandler(pc.consumerID, pc)
+	if err != nil {
+		pc.log.WithError(err).Error("Failed to add consumer handler")
+		return err
+	}
 
 	msgType := res.Response.GetType()
 
diff --git a/pulsar/internal/auth/basic.go b/pulsar/internal/auth/basic.go
new file mode 100644
index 0000000..58a87f5
--- /dev/null
+++ b/pulsar/internal/auth/basic.go
@@ -0,0 +1,84 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package auth
+
+import (
+	"crypto/tls"
+	"encoding/base64"
+	"errors"
+	"net/http"
+)
+
+type basicAuthProvider struct {
+	rt               http.RoundTripper
+	commandAuthToken []byte
+	httpAuthToken    string
+}
+
+func NewAuthenticationBasic(username, password string) (Provider, error) {
+	if username == "" {
+		return nil, errors.New("username cannot be empty")
+	}
+	if password == "" {
+		return nil, errors.New("password cannot be empty")
+	}
+
+	commandAuthToken := []byte(username + ":" + password)
+	return &basicAuthProvider{
+		commandAuthToken: commandAuthToken,
+		httpAuthToken:    "Basic " + base64.StdEncoding.EncodeToString(commandAuthToken),
+	}, nil
+}
+
+func NewAuthenticationBasicWithParams(params map[string]string) (Provider, error) {
+	return NewAuthenticationBasic(params["username"], params["password"])
+}
+
+func (b *basicAuthProvider) Init() error {
+	return nil
+}
+
+func (b *basicAuthProvider) Name() string {
+	return "basic"
+}
+
+func (b *basicAuthProvider) GetTLSCertificate() (*tls.Certificate, error) {
+	return nil, nil
+}
+
+func (b *basicAuthProvider) GetData() ([]byte, error) {
+	return b.commandAuthToken, nil
+}
+
+func (b *basicAuthProvider) Close() error {
+	return nil
+}
+
+func (b *basicAuthProvider) RoundTrip(req *http.Request) (*http.Response, error) {
+	req.Header.Add("Authorization", b.httpAuthToken)
+	return b.rt.RoundTrip(req)
+}
+
+func (b *basicAuthProvider) Transport() http.RoundTripper {
+	return b.rt
+}
+
+func (b *basicAuthProvider) WithTransport(tr http.RoundTripper) error {
+	b.rt = tr
+	return nil
+}
diff --git a/pulsar/internal/auth/provider.go b/pulsar/internal/auth/provider.go
index 1731490..031ea8d 100644
--- a/pulsar/internal/auth/provider.go
+++ b/pulsar/internal/auth/provider.go
@@ -80,6 +80,9 @@ func NewProvider(name string, params string) (Provider, error) {
 	case "oauth2", "org.apache.pulsar.client.impl.auth.oauth2.AuthenticationOAuth2":
 		return NewAuthenticationOAuth2WithParams(m)
 
+	case "basic", "org.apache.pulsar.client.impl.auth.AuthenticationBasic":
+		return NewAuthenticationBasicWithParams(m)
+
 	default:
 		return nil, fmt.Errorf("invalid auth provider '%s'", name)
 	}
diff --git a/pulsar/internal/batch_builder.go b/pulsar/internal/batch_builder.go
index 9d18f26..fb7598e 100644
--- a/pulsar/internal/batch_builder.go
+++ b/pulsar/internal/batch_builder.go
@@ -157,6 +157,10 @@ func (bc *batchContainer) IsFull() bool {
 
 // hasSpace should return true if and only if the batch container can accommodate another message of length payload.
 func (bc *batchContainer) hasSpace(payload []byte) bool {
+	if bc.numMessages == 0 {
+		// allow to add at least one message when batching is disabled
+		return true
+	}
 	msgSize := uint32(len(payload))
 	return bc.numMessages+1 <= bc.maxMessages && bc.buffer.ReadableBytes()+msgSize <= uint32(bc.maxBatchSize)
 }
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index fa8d055..0249020 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -53,8 +53,9 @@ type TLSOptions struct {
 }
 
 var (
-	errConnectionClosed       = errors.New("connection closed")
-	errUnableRegisterListener = errors.New("unable register listener when con closed")
+	errConnectionClosed        = errors.New("connection closed")
+	errUnableRegisterListener  = errors.New("unable register listener when con closed")
+	errUnableAddConsumeHandler = errors.New("unable add consumer handler when con closed")
 )
 
 // ConnectionListener is a user of a connection (eg. a producer or
@@ -75,7 +76,7 @@ type Connection interface {
 	WriteData(data Buffer)
 	RegisterListener(id uint64, listener ConnectionListener) error
 	UnregisterListener(id uint64)
-	AddConsumeHandler(id uint64, handler ConsumerHandler)
+	AddConsumeHandler(id uint64, handler ConsumerHandler) error
 	DeleteConsumeHandler(id uint64)
 	ID() string
 	GetMaxMessageSize() int32
@@ -994,16 +995,17 @@ func (c *connection) getTLSConfig() (*tls.Config, error) {
 	return tlsConfig, nil
 }
 
-func (c *connection) AddConsumeHandler(id uint64, handler ConsumerHandler) {
+func (c *connection) AddConsumeHandler(id uint64, handler ConsumerHandler) error {
 	// do not add if connection is closed
 	if c.closed() {
 		c.log.Warnf("Closed connection unable add consumer with id=%+v", id)
-		return
+		return errUnableAddConsumeHandler
 	}
 
 	c.consumerHandlersLock.Lock()
 	defer c.consumerHandlersLock.Unlock()
 	c.consumerHandlers[id] = handler
+	return nil
 }
 
 func (c *connection) DeleteConsumeHandler(id uint64) {
diff --git a/pulsar/internal/key_based_batch_builder.go b/pulsar/internal/key_based_batch_builder.go
index d09138c..667e855 100644
--- a/pulsar/internal/key_based_batch_builder.go
+++ b/pulsar/internal/key_based_batch_builder.go
@@ -117,6 +117,11 @@ func (bc *keyBasedBatchContainer) IsMultiBatches() bool {
 
 // hasSpace should return true if and only if the batch container can accommodate another message of length payload.
 func (bc *keyBasedBatchContainer) hasSpace(payload []byte) bool {
+	if bc.numMessages == 0 {
+		// allow to add at least one message
+		// and a single max message size is checked in the producer partition, therefore no need to validate batch size
+		return true
+	}
 	msgSize := uint32(len(payload))
 	return bc.numMessages+1 <= bc.maxMessages && bc.buffer.ReadableBytes()+msgSize <= uint32(bc.maxBatchSize)
 }
diff --git a/pulsar/schema.go b/pulsar/schema.go
index 8021ffe..42499ad 100644
--- a/pulsar/schema.go
+++ b/pulsar/schema.go
@@ -91,18 +91,29 @@ type JSONSchema struct {
 	SchemaInfo
 }
 
+// NewJSONSchema creates a new JSONSchema
+// Note: the function will panic if creation of codec fails
 func NewJSONSchema(jsonAvroSchemaDef string, properties map[string]string) *JSONSchema {
+	js, err := NewJSONSchemaWithValidation(jsonAvroSchemaDef, properties)
+	if err != nil {
+		log.Fatalf("JSONSchema init codec error:%v", err)
+	}
+	return js
+}
+
+// NewJSONSchemaWithValidation creates a new JSONSchema and error to indicate codec failure
+func NewJSONSchemaWithValidation(jsonAvroSchemaDef string, properties map[string]string) (*JSONSchema, error) {
 	js := new(JSONSchema)
 	avroCodec, err := initAvroCodec(jsonAvroSchemaDef)
 	if err != nil {
-		log.Fatalf("init codec error:%v", err)
+		return nil, err
 	}
 	schemaDef := NewSchemaDefinition(avroCodec)
 	js.SchemaInfo.Schema = schemaDef.Codec.Schema()
 	js.SchemaInfo.Type = JSON
 	js.SchemaInfo.Properties = properties
 	js.SchemaInfo.Name = "JSON"
-	return js
+	return js, nil
 }
 
 func (js *JSONSchema) Encode(data interface{}) ([]byte, error) {
@@ -126,11 +137,22 @@ type ProtoSchema struct {
 	SchemaInfo
 }
 
+// NewProtoSchema creates a new ProtoSchema
+// Note: the function will panic if creation of codec fails
 func NewProtoSchema(protoAvroSchemaDef string, properties map[string]string) *ProtoSchema {
+	ps, err := NewProtoSchemaWithValidation(protoAvroSchemaDef, properties)
+	if err != nil {
+		log.Fatalf("ProtoSchema init codec error:%v", err)
+	}
+	return ps
+}
+
+// NewProtoSchemaWithValidation creates a new ProtoSchema and error to indicate codec failure
+func NewProtoSchemaWithValidation(protoAvroSchemaDef string, properties map[string]string) (*ProtoSchema, error) {
 	ps := new(ProtoSchema)
 	avroCodec, err := initAvroCodec(protoAvroSchemaDef)
 	if err != nil {
-		log.Fatalf("init codec error:%v", err)
+		return nil, err
 	}
 	schemaDef := NewSchemaDefinition(avroCodec)
 	ps.AvroCodec.Codec = schemaDef.Codec
@@ -138,7 +160,7 @@ func NewProtoSchema(protoAvroSchemaDef string, properties map[string]string) *Pr
 	ps.SchemaInfo.Type = PROTOBUF
 	ps.SchemaInfo.Properties = properties
 	ps.SchemaInfo.Name = "Proto"
-	return ps
+	return ps, nil
 }
 
 func (ps *ProtoSchema) Encode(data interface{}) ([]byte, error) {
@@ -162,11 +184,22 @@ type AvroSchema struct {
 	SchemaInfo
 }
 
+// NewAvroSchema creates a new AvroSchema
+// Note: the function will panic if creation of codec fails
 func NewAvroSchema(avroSchemaDef string, properties map[string]string) *AvroSchema {
+	ps, err := NewAvroSchemaWithValidation(avroSchemaDef, properties)
+	if err != nil {
+		log.Fatalf("AvroSchema init codec error:%v", err)
+	}
+	return ps
+}
+
+// NewAvroSchemaWithValidation creates a new AvroSchema and error to indicate codec failure
+func NewAvroSchemaWithValidation(avroSchemaDef string, properties map[string]string) (*AvroSchema, error) {
 	as := new(AvroSchema)
 	avroCodec, err := initAvroCodec(avroSchemaDef)
 	if err != nil {
-		log.Fatalf("init codec error:%v", err)
+		return nil, err
 	}
 	schemaDef := NewSchemaDefinition(avroCodec)
 	as.AvroCodec.Codec = schemaDef.Codec
@@ -174,7 +207,7 @@ func NewAvroSchema(avroSchemaDef string, properties map[string]string) *AvroSche
 	as.SchemaInfo.Type = AVRO
 	as.SchemaInfo.Name = "Avro"
 	as.SchemaInfo.Properties = properties
-	return as
+	return as, nil
 }
 
 func (as *AvroSchema) Encode(data interface{}) ([]byte, error) {
@@ -235,14 +268,10 @@ func (ss *StringSchema) Encode(v interface{}) ([]byte, error) {
 	return []byte(v.(string)), nil
 }
 
+// Decode convert from byte slice to string without allocating a new string
 func (ss *StringSchema) Decode(data []byte, v interface{}) error {
-	bh := (*reflect.SliceHeader)(unsafe.Pointer(&data))
-	sh := reflect.StringHeader{
-		Data: bh.Data,
-		Len:  bh.Len,
-	}
-	shPtr := (*string)(unsafe.Pointer(&sh))
-	reflect.ValueOf(v).Elem().Set(reflect.ValueOf(shPtr))
+	strPtr := (*string)(unsafe.Pointer(&data))
+	reflect.ValueOf(v).Elem().Set(reflect.ValueOf(strPtr))
 	return nil
 }
 


Test patch:
diff --git a/integration-tests/.htpasswd b/integration-tests/.htpasswd
new file mode 100644
index 0000000..2aa3a47
--- /dev/null
+++ b/integration-tests/.htpasswd
@@ -0,0 +1,1 @@
+admin:$apr1$FG4AO6aX$KGYPuMoLUou3i6vUkPUUf.
diff --git a/integration-tests/license_test.go b/integration-tests/license_test.go
index e829560..cd6d9d5 100644
--- a/integration-tests/license_test.go
+++ b/integration-tests/license_test.go
@@ -68,6 +68,7 @@ var skip = map[string]bool{
 	"../pulsar/internal/pulsar_proto/PulsarApi.pb.go": true,
 	"../.github/workflows/bot.yaml":                   true,
 	"../integration-tests/pb/hello.pb.go":             true,
+	"../integration-tests/.htpasswd":                  true,
 }
 
 func TestLicense(t *testing.T) {
diff --git a/integration-tests/standalone.conf b/integration-tests/standalone.conf
index a298a61..8cd2828 100644
--- a/integration-tests/standalone.conf
+++ b/integration-tests/standalone.conf
@@ -98,7 +98,7 @@ anonymousUserRole=anonymous
 authenticationEnabled=true
 
 # Autentication provider name list, which is comma separated list of class names
-authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderTls,org.apache.pulsar.broker.authentication.AuthenticationProviderToken
+authenticationProviders=org.apache.pulsar.broker.authentication.AuthenticationProviderTls,org.apache.pulsar.broker.authentication.AuthenticationProviderToken,org.apache.pulsar.broker.authentication.AuthenticationProviderBasic
 
 # Enforce authorization
 authorizationEnabled=true
diff --git a/pulsar/client_impl_test.go b/pulsar/client_impl_test.go
index 2d83c99..ba8f6eb 100644
--- a/pulsar/client_impl_test.go
+++ b/pulsar/client_impl_test.go
@@ -28,6 +28,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/stretchr/testify/require"
+
 	"github.com/apache/pulsar-client-go/pulsar/internal"
 
 	"github.com/apache/pulsar-client-go/pulsar/internal/auth"
@@ -1019,3 +1021,48 @@ func TestHTTPOAuth2AuthFailed(t *testing.T) {
 
 	client.Close()
 }
+
+func TestHTTPBasicAuth(t *testing.T) {
+	basicAuth, err := NewAuthenticationBasic("admin", "123456")
+	require.NoError(t, err)
+	require.NotNil(t, basicAuth)
+
+	client, err := NewClient(ClientOptions{
+		URL:            webServiceURL,
+		Authentication: basicAuth,
+	})
+	require.NoError(t, err)
+	require.NotNil(t, client)
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: newAuthTopicName(),
+	})
+
+	require.NoError(t, err)
+	require.NotNil(t, producer)
+
+	client.Close()
+}
+
+func TestHTTPSBasicAuth(t *testing.T) {
+	basicAuth, err := NewAuthenticationBasic("admin", "123456")
+	require.NoError(t, err)
+	require.NotNil(t, basicAuth)
+
+	client, err := NewClient(ClientOptions{
+		URL:                   webServiceURLTLS,
+		TLSTrustCertsFilePath: caCertsPath,
+		TLSValidateHostname:   true,
+		Authentication:        basicAuth,
+	})
+	require.NoError(t, err)
+	require.NotNil(t, client)
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic: newAuthTopicName(),
+	})
+	require.NoError(t, err)
+	require.NotNil(t, producer)
+
+	client.Close()
+}
diff --git a/pulsar/internal/auth/basic_test.go b/pulsar/internal/auth/basic_test.go
new file mode 100644
index 0000000..b212d0b
--- /dev/null
+++ b/pulsar/internal/auth/basic_test.go
@@ -0,0 +1,70 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package auth
+
+import (
+	"errors"
+	"io/ioutil"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewAuthenticationBasicWithParams(t *testing.T) {
+	username := "admin"
+	password := "123456"
+
+	provider, err := NewAuthenticationBasic(username, password)
+	require.NoError(t, err)
+	require.NotNil(t, provider)
+
+	data, err := provider.GetData()
+	require.NoError(t, err)
+	require.Equal(t, []byte(username+":"+password), data)
+
+	s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		_, _ = w.Write([]byte(r.Header.Get("Authorization")))
+	}))
+
+	client := s.Client()
+	err = provider.WithTransport(client.Transport)
+	require.NoError(t, err)
+	client.Transport = provider
+
+	resp, err := client.Get(s.URL)
+	require.NoError(t, err)
+
+	body, err := ioutil.ReadAll(resp.Body)
+	_ = resp.Body.Close()
+	require.NoError(t, err)
+	require.Equal(t, []byte("Basic YWRtaW46MTIzNDU2"), body)
+}
+
+func TestNewAuthenticationBasicWithInvalidParams(t *testing.T) {
+	username := "admin"
+	password := "123456"
+	provider, err := NewAuthenticationBasic("", password)
+	require.Equal(t, errors.New("username cannot be empty"), err)
+	require.Nil(t, provider)
+
+	provider, err = NewAuthenticationBasic(username, "")
+	require.Equal(t, errors.New("password cannot be empty"), err)
+	require.Nil(t, provider)
+}
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index 541c1fe..6b2b5d9 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -877,20 +877,44 @@ func TestMaxBatchSize(t *testing.T) {
 	assert.NotNil(t, producer)
 	defer producer.Close()
 
-	for bias := -1; bias <= 1; bias++ {
+	for bias := -1; bias <= 3; bias++ {
 		payload := make([]byte, batchMaxMessageSize+bias)
 		ID, err := producer.Send(context.Background(), &ProducerMessage{
 			Payload: payload,
 		})
-		if bias <= 0 {
-			assert.NoError(t, err)
-			assert.NotNil(t, ID)
-		} else {
-			assert.Equal(t, errFailAddToBatch, err)
-		}
+		// regardless max batch size, if the batch size limit is reached, batching is triggered to send messages
+		assert.NoError(t, err)
+		assert.NotNil(t, ID)
 	}
 }
 
+func TestBatchingDisabled(t *testing.T) {
+	defaultMaxMessageSize := 128 * 1024
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+
+	// when batching is disabled, the batching size has no effect
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:           newTopicName(),
+		DisableBatching: true,
+		BatchingMaxSize: uint(defaultMaxBatchSize),
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, producer)
+	defer producer.Close()
+
+	payload := make([]byte, defaultMaxMessageSize+100)
+	ID, err := producer.Send(context.Background(), &ProducerMessage{
+		Payload: payload,
+	})
+	// regardless max batch size, if the batch size limit is reached, batching is triggered to send messages
+	assert.NoError(t, err)
+	assert.NotNil(t, ID)
+}
+
 func TestMaxMessageSize(t *testing.T) {
 	serverMaxMessageSize := 1024 * 1024
 
diff --git a/pulsar/schema_test.go b/pulsar/schema_test.go
index ce19bac..1aa35f0 100644
--- a/pulsar/schema_test.go
+++ b/pulsar/schema_test.go
@@ -82,7 +82,9 @@ func TestJsonSchema(t *testing.T) {
 	//create consumer
 	var s testJSON
 
-	consumerJS := NewJSONSchema(exampleSchemaDef, nil)
+	consumerJS, err := NewJSONSchemaWithValidation(exampleSchemaDef, nil)
+	assert.Nil(t, err)
+	consumerJS = NewJSONSchema(exampleSchemaDef, nil) // test this legacy function
 	consumer, err := client.Subscribe(ConsumerOptions{
 		Topic:                       "jsonTopic",
 		SubscriptionName:            "sub-1",
@@ -105,7 +107,9 @@ func TestProtoSchema(t *testing.T) {
 	defer client.Close()
 
 	// create producer
-	psProducer := NewProtoSchema(protoSchemaDef, nil)
+	psProducer, err := NewProtoSchemaWithValidation(protoSchemaDef, nil)
+	assert.Nil(t, err)
+	psProducer = NewProtoSchema(protoSchemaDef, nil)
 	producer, err := client.CreateProducer(ProducerOptions{
 		Topic:  "proto",
 		Schema: psProducer,
@@ -146,7 +150,9 @@ func TestAvroSchema(t *testing.T) {
 	defer client.Close()
 
 	// create producer
-	asProducer := NewAvroSchema(exampleSchemaDef, nil)
+	asProducer, err := NewAvroSchemaWithValidation(exampleSchemaDef, nil)
+	assert.Nil(t, err)
+	asProducer = NewAvroSchema(exampleSchemaDef, nil)
 	producer, err := client.CreateProducer(ProducerOptions{
 		Topic:  "avro-topic",
 		Schema: asProducer,