Repository: apache/pulsar-client-go
Pull request #886 (2022-11-03 09:16:20)

prevent consumer panic on de-serializing message if schema not found
address issue #884 


Patch:
diff --git a/.github/PULL_REQUEST_TEMPLATE.md b/.github/PULL_REQUEST_TEMPLATE.md
index d78c634..a9853d0 100644
--- a/.github/PULL_REQUEST_TEMPLATE.md
+++ b/.github/PULL_REQUEST_TEMPLATE.md
@@ -1,4 +1,4 @@
-<--
+<!--
 ### Contribution Checklist
   
   - Name the pull request in the form "[Issue XYZ][component] Title of the pull request", where *XYZ* should be replaced by the actual issue number.
diff --git a/oauth2/client_credentials_provider.go b/oauth2/client_credentials_provider.go
index 78c7228..5230ca3 100644
--- a/oauth2/client_credentials_provider.go
+++ b/oauth2/client_credentials_provider.go
@@ -55,8 +55,14 @@ func (k *KeyFileProvider) GetClientCredentials() (*KeyFile, error) {
 	case strings.HasPrefix(k.KeyFile, FILE):
 		filename := strings.TrimPrefix(k.KeyFile, FILE)
 		keyFile, err = ioutil.ReadFile(filename)
-	case strings.HasPrefix(k.KeyFile, "data://"):
+	case strings.HasPrefix(k.KeyFile, DATA):
 		keyFile = []byte(strings.TrimPrefix(k.KeyFile, DATA))
+	case strings.HasPrefix(k.KeyFile, "data:"):
+		url, err := newDataURL(k.KeyFile)
+		if err != nil {
+			return nil, err
+		}
+		keyFile = url.Data
 	default:
 		keyFile, err = ioutil.ReadFile(k.KeyFile)
 	}
diff --git a/oauth2/data_url.go b/oauth2/data_url.go
new file mode 100644
index 0000000..8d711f8
--- /dev/null
+++ b/oauth2/data_url.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 oauth2
+
+import (
+	"encoding/base64"
+	"errors"
+	"regexp"
+)
+
+var errDataURLInvalid = errors.New("invalid data URL")
+
+// https://datatracker.ietf.org/doc/html/rfc2397
+var dataURLRegex = regexp.MustCompile("^data:(?P<mimetype>[^;,]+)?(;(?P<charset>charset=[^;,]+))?" +
+	"(;(?P<base64>base64))?,(?P<data>.+)")
+
+type dataURL struct {
+	url      string
+	Mimetype string
+	Data     []byte
+}
+
+func newDataURL(url string) (*dataURL, error) {
+	if !dataURLRegex.Match([]byte(url)) {
+		return nil, errDataURLInvalid
+	}
+
+	match := dataURLRegex.FindStringSubmatch(url)
+	if len(match) != 7 {
+		return nil, errDataURLInvalid
+	}
+
+	dataURL := &dataURL{
+		url: url,
+	}
+
+	mimetype := match[dataURLRegex.SubexpIndex("mimetype")]
+	if mimetype == "" {
+		mimetype = "text/plain"
+	}
+	dataURL.Mimetype = mimetype
+
+	data := match[dataURLRegex.SubexpIndex("data")]
+	if match[dataURLRegex.SubexpIndex("base64")] == "" {
+		dataURL.Data = []byte(data)
+	} else {
+		data, err := base64.StdEncoding.DecodeString(data)
+		if err != nil {
+			return nil, err
+		}
+		dataURL.Data = data
+	}
+
+	return dataURL, nil
+}
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 1ade1e2..4149a91 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -221,6 +221,11 @@ func (s *schemaInfoCache) Get(schemaVersion []byte) (schema Schema, err error) {
 		return nil, err
 	}
 
+	if pbSchema == nil {
+		err = fmt.Errorf("schema not found for topic: [ %v ], schema version : [ %v ]", s.topic, schemaVersion)
+		return nil, err
+	}
+
 	var properties = internal.ConvertToStringMap(pbSchema.Properties)
 
 	schema, err = NewSchema(SchemaType(*pbSchema.Type), pbSchema.SchemaData, properties)
@@ -469,7 +474,7 @@ func (pc *partitionConsumer) AckID(msgID MessageID) error {
 		pc.options.interceptors.OnAcknowledge(pc.parentConsumer, msgID)
 	}
 
-	return ackReq.err
+	return nil
 }
 
 func (pc *partitionConsumer) NackID(msgID MessageID) {
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index c34f25d..fc1b349 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -986,7 +986,7 @@ func (c *connection) getTLSConfig() (*tls.Config, error) {
 		c.log.Debugf("getTLSConfig(): setting tlsConfig.ServerName = %+v", tlsConfig.ServerName)
 	}
 
-	if c.tlsOptions.CertFile != "" || c.tlsOptions.KeyFile != "" {
+	if c.tlsOptions.CertFile != "" && c.tlsOptions.KeyFile != "" {
 		cert, err := tls.LoadX509KeyPair(c.tlsOptions.CertFile, c.tlsOptions.KeyFile)
 		if err != nil {
 			return nil, errors.New(err.Error())
diff --git a/pulsar/internal/http_client.go b/pulsar/internal/http_client.go
index 3c68dac..7abf70d 100644
--- a/pulsar/internal/http_client.go
+++ b/pulsar/internal/http_client.go
@@ -346,7 +346,7 @@ func getDefaultTransport(tlsConfig *TLSOptions) (http.RoundTripper, error) {
 			cfg.RootCAs.AppendCertsFromPEM(rootCA)
 		}
 
-		if tlsConfig.CertFile != "" || tlsConfig.KeyFile != "" {
+		if tlsConfig.CertFile != "" && tlsConfig.KeyFile != "" {
 			cert, err := tls.LoadX509KeyPair(tlsConfig.CertFile, tlsConfig.KeyFile)
 			if err != nil {
 				return nil, errors.New(err.Error())
diff --git a/pulsar/reader.go b/pulsar/reader.go
index 3539037..e4679ab 100644
--- a/pulsar/reader.go
+++ b/pulsar/reader.go
@@ -20,6 +20,8 @@ package pulsar
 import (
 	"context"
 	"time"
+
+	"github.com/apache/pulsar-client-go/pulsar/internal"
 )
 
 // ReaderMessage packages Reader and Message as a struct to use.
@@ -86,6 +88,10 @@ type ReaderOptions struct {
 
 	// Schema represents the schema implementation.
 	Schema Schema
+
+	// BackoffPolicy parameterize the following options in the reconnection logic to
+	// allow users to customize the reconnection logic (minBackoff, maxBackoff and jitterPercentage)
+	BackoffPolicy internal.BackoffPolicy
 }
 
 // Reader can be used to scan through all the messages currently available in a topic.
diff --git a/pulsar/reader_impl.go b/pulsar/reader_impl.go
index dd552b6..3931f36 100644
--- a/pulsar/reader_impl.go
+++ b/pulsar/reader_impl.go
@@ -106,6 +106,7 @@ func newReader(client *client, options ReaderOptions) (Reader, error) {
 		replicateSubscriptionState: false,
 		decryption:                 options.Decryption,
 		schema:                     options.Schema,
+		backoffPolicy:              options.BackoffPolicy,
 	}
 
 	reader := &reader{


Test patch:
diff --git a/oauth2/client_credentials_provider_test.go b/oauth2/client_credentials_provider_test.go
new file mode 100644
index 0000000..f47967b
--- /dev/null
+++ b/oauth2/client_credentials_provider_test.go
@@ -0,0 +1,78 @@
+// 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 oauth2
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"io/ioutil"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewClientCredentialsProviderFromKeyFile(t *testing.T) {
+	oauthType := "TYPE"
+	clientID := "CLIENT_ID"
+	ClientSecret := "CLIENT_SECRET"
+	ClientEmail := "CLIENT_EMAIL"
+	IssuerURL := "ISSUER_URL"
+	keyFile := &KeyFile{
+		Type:         oauthType,
+		ClientID:     clientID,
+		ClientSecret: ClientSecret,
+		ClientEmail:  ClientEmail,
+		IssuerURL:    IssuerURL,
+	}
+
+	b, err := json.Marshal(keyFile)
+	require.NoError(t, err)
+	tmpFile, err := ioutil.TempFile("", "key-file")
+	require.NoError(t, err)
+	defer func(name string) {
+		_ = os.Remove(name)
+	}(tmpFile.Name())
+	_, err = tmpFile.Write(b)
+	require.NoError(t, err)
+
+	jsonData := string(b)
+	base64Data := base64.StdEncoding.EncodeToString(b)
+
+	assertCredentials(t, fmt.Sprintf("file://%s", tmpFile.Name()), keyFile)
+	assertCredentials(t, fmt.Sprintf("data://%s", jsonData), keyFile)
+	assertCredentials(t, fmt.Sprintf("data:,%s", jsonData), keyFile)
+	assertCredentials(t, fmt.Sprintf("data:application/json,%s", jsonData), keyFile)
+	assertCredentials(t, fmt.Sprintf("data:;base64,%s", base64Data), keyFile)
+	assertCredentials(t, fmt.Sprintf("data:application/json;base64,%s", base64Data), keyFile)
+}
+
+func TestNewInvalidClientCredentialsProviderFromKeyFile(t *testing.T) {
+	p := NewClientCredentialsProviderFromKeyFile("data:application/data,hi")
+	_, err := p.GetClientCredentials()
+	require.Error(t, err)
+}
+
+func assertCredentials(t *testing.T, keyfile string, expected *KeyFile) {
+	p := NewClientCredentialsProviderFromKeyFile(keyfile)
+	clientCredentials, err := p.GetClientCredentials()
+	require.NoError(t, err)
+	assert.Equal(t, expected, clientCredentials)
+}
diff --git a/oauth2/data_url_test.go b/oauth2/data_url_test.go
new file mode 100644
index 0000000..8bdc880
--- /dev/null
+++ b/oauth2/data_url_test.go
@@ -0,0 +1,58 @@
+// 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 oauth2
+
+import (
+	"encoding/base64"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewDataURL(t *testing.T) {
+	rawURL := "data:,test"
+	url, err := newDataURL(rawURL)
+	require.NoError(t, err)
+	assert.Equal(t, "text/plain", url.Mimetype)
+	assert.Equal(t, "test", string(url.Data))
+
+	rawURL = "data:;base64," + base64.StdEncoding.EncodeToString([]byte("test"))
+	url, err = newDataURL(rawURL)
+	require.NoError(t, err)
+	assert.Equal(t, "text/plain", url.Mimetype)
+	assert.Equal(t, "test", string(url.Data))
+
+	rawURL = "data:application/json,test"
+	url, err = newDataURL(rawURL)
+	require.NoError(t, err)
+	assert.Equal(t, "application/json", url.Mimetype)
+	assert.Equal(t, "test", string(url.Data))
+
+	rawURL = "data:application/json;base64," + base64.StdEncoding.EncodeToString([]byte("test"))
+	url, err = newDataURL(rawURL)
+	require.NoError(t, err)
+	assert.Equal(t, "application/json", url.Mimetype)
+	assert.Equal(t, "test", string(url.Data))
+
+	rawURL = "data://test"
+	url, err = newDataURL(rawURL)
+	require.Nil(t, url)
+	assert.Error(t, err)
+	assert.EqualError(t, errDataURLInvalid, err.Error())
+}
diff --git a/pulsar/consumer_test.go b/pulsar/consumer_test.go
index f574378..e9b3013 100644
--- a/pulsar/consumer_test.go
+++ b/pulsar/consumer_test.go
@@ -3262,3 +3262,44 @@ func TestAvailablePermitsLeak(t *testing.T) {
 	assert.NotEqual(t, true, errors.Is(err, context.DeadlineExceeded),
 		"This means the resource is exhausted. consumer.Receive() will block forever.")
 }
+
+func TestConsumerWithBackoffPolicy(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+
+	topicName := newTopicName()
+
+	backoff := newTestBackoffPolicy(1*time.Second, 4*time.Second)
+	_consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:            topicName,
+		SubscriptionName: "sub-1",
+		Type:             Shared,
+		BackoffPolicy:    backoff,
+	})
+	assert.Nil(t, err)
+	defer _consumer.Close()
+
+	partitionConsumerImp := _consumer.(*consumer).consumers[0]
+	// 1 s
+	startTime := time.Now()
+	partitionConsumerImp.reconnectToBroker()
+	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+
+	// 2 s
+	startTime = time.Now()
+	partitionConsumerImp.reconnectToBroker()
+	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+
+	// 4 s
+	startTime = time.Now()
+	partitionConsumerImp.reconnectToBroker()
+	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+
+	// 4 s
+	startTime = time.Now()
+	partitionConsumerImp.reconnectToBroker()
+	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+}
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index f193ffd..75fc6db 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -1072,6 +1072,46 @@ func TestSendTimeout(t *testing.T) {
 	makeHTTPCall(t, http.MethodDelete, quotaURL, "")
 }
 
+func TestProducerWithBackoffPolicy(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+
+	topicName := newTopicName()
+
+	backoff := newTestBackoffPolicy(1*time.Second, 4*time.Second)
+	_producer, err := client.CreateProducer(ProducerOptions{
+		Topic:         topicName,
+		SendTimeout:   2 * time.Second,
+		BackoffPolicy: backoff,
+	})
+	assert.Nil(t, err)
+	defer _producer.Close()
+
+	partitionProducerImp := _producer.(*producer).producers[0].(*partitionProducer)
+	// 1 s
+	startTime := time.Now()
+	partitionProducerImp.reconnectToBroker()
+	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+
+	// 2 s
+	startTime = time.Now()
+	partitionProducerImp.reconnectToBroker()
+	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+
+	// 4 s
+	startTime = time.Now()
+	partitionProducerImp.reconnectToBroker()
+	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+
+	// 4 s
+	startTime = time.Now()
+	partitionProducerImp.reconnectToBroker()
+	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+}
+
 func TestSendContextExpired(t *testing.T) {
 	quotaURL := adminURL + "/admin/v2/namespaces/public/default/backlogQuota"
 	quotaFmt := `{"limit": "%d", "policy": "producer_request_hold"}`
@@ -1600,3 +1640,68 @@ func TestMultipleSchemaProducerConsumer(t *testing.T) {
 		assert.Nil(t, err)
 	}
 }
+
+func TestProducerWithSchemaAndConsumerSchemaNotFound(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: lookupURL,
+	})
+	assert.NoError(t, err)
+	defer client.Close()
+
+	schema := NewAvroSchema(`{"fields":
+	[
+		{"name":"id","type":"int"},{"default":null,"name":"name","type":["null","string"]}
+	],
+	"name":"MyAvro","namespace":"schemaNotFoundTestCase","type":"record"}`, nil)
+
+	topic := newTopicName()
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:  topic,
+		Schema: schema,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, producer)
+	defer producer.Close()
+
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:                       topic,
+		SubscriptionName:            "my-sub-schema-not-found",
+		Type:                        Exclusive,
+		Schema:                      schema,
+		SubscriptionInitialPosition: SubscriptionPositionEarliest,
+	})
+	assert.NoError(t, err)
+	assert.NotNil(t, consumer)
+
+	// each produced message will have schema version set in the message metadata
+	for i := 0; i < 5; i++ {
+		messageContent, err := schema.Encode(map[string]interface{}{
+			"id": i,
+			"name": map[string]interface{}{
+				"string": "abc",
+			},
+		})
+		assert.NoError(t, err)
+		_, err = producer.Send(context.Background(), &ProducerMessage{
+			Payload: messageContent,
+		})
+		assert.NoError(t, err)
+	}
+
+	// delete schema of topic
+	topicSchemaDeleteURL := fmt.Sprintf("admin/v2/schemas/public/default/%v/schema", topic)
+	err = httpDelete(topicSchemaDeleteURL)
+	assert.NoError(t, err)
+
+	// consume message
+	msg, err := consumer.Receive(context.Background())
+	assert.NoError(t, err)
+	assert.NotNil(t, msg)
+
+	// try to serialize message payload
+	var v interface{}
+	err = msg.GetSchemaValue(&v)
+	// should fail with error but not panic
+	assert.Error(t, err)
+}
diff --git a/pulsar/reader_test.go b/pulsar/reader_test.go
index 3c85871..07e7ed3 100644
--- a/pulsar/reader_test.go
+++ b/pulsar/reader_test.go
@@ -775,3 +775,78 @@ func TestReaderWithSchema(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, *res, value)
 }
+
+func newTestBackoffPolicy(minBackoff, maxBackoff time.Duration) *testBackoffPolicy {
+	return &testBackoffPolicy{
+		curBackoff: 0,
+		minBackoff: minBackoff,
+		maxBackoff: maxBackoff,
+	}
+}
+
+type testBackoffPolicy struct {
+	curBackoff, minBackoff, maxBackoff time.Duration
+	retryTime                          int
+}
+
+func (b *testBackoffPolicy) Next() time.Duration {
+	// Double the delay each time
+	b.curBackoff += b.curBackoff
+	if b.curBackoff.Nanoseconds() < b.minBackoff.Nanoseconds() {
+		b.curBackoff = b.minBackoff
+	} else if b.curBackoff.Nanoseconds() > b.maxBackoff.Nanoseconds() {
+		b.curBackoff = b.maxBackoff
+	}
+	b.retryTime++
+
+	return b.curBackoff
+}
+
+func (b *testBackoffPolicy) IsExpectedIntervalFrom(startTime time.Time) bool {
+	// Approximately equal to expected interval
+	if time.Since(startTime) < b.curBackoff-time.Second {
+		return false
+	}
+	if time.Since(startTime) > b.curBackoff+time.Second {
+		return false
+	}
+	return true
+}
+
+func TestReaderWithBackoffPolicy(t *testing.T) {
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+	assert.Nil(t, err)
+	defer client.Close()
+
+	backoff := newTestBackoffPolicy(1*time.Second, 4*time.Second)
+	_reader, err := client.CreateReader(ReaderOptions{
+		Topic:          "my-topic",
+		StartMessageID: LatestMessageID(),
+		BackoffPolicy:  backoff,
+	})
+	assert.NotNil(t, _reader)
+	assert.Nil(t, err)
+
+	partitionConsumerImp := _reader.(*reader).pc
+	// 1 s
+	startTime := time.Now()
+	partitionConsumerImp.reconnectToBroker()
+	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+
+	// 2 s
+	startTime = time.Now()
+	partitionConsumerImp.reconnectToBroker()
+	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+
+	// 4 s
+	startTime = time.Now()
+	partitionConsumerImp.reconnectToBroker()
+	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+
+	// 4 s
+	startTime = time.Now()
+	partitionConsumerImp.reconnectToBroker()
+	assert.True(t, backoff.IsExpectedIntervalFrom(startTime))
+}