Repository: apache/pulsar-client-go
Pull request #881 (2022-11-02 07:48:38)

[Fix] Fix the data race of ackReq.err
Master Issue: #879 

### Motivation

#879 shows ackReq.err has data race problem.

Read
https://github.com/apache/pulsar-client-go/blob/44b64aa712a0d85ce7562134c0f7cf3d21f17ac0/pulsar/consumer_partition.go#L471-L473

Write
https://github.com/apache/pulsar-client-go/blob/44b64aa712a0d85ce7562134c0f7cf3d21f17ac0/pulsar/consumer_partition.go#L702-L707

After ackReq.doneCh has introduced in #777 , I think the final return value of `AckId` can be nil. Because ` ackReq.err` may not necessarily be set before the `AckID` returns. If user need to know the result of ack, just using `AckIDWithResponse` will be fine.

### Modifications

Modify the return value of `AckID`.

### Verifying this change

- [x] Make sure that the change passes the CI checks.

*(Please pick either of the following options)*

This change is a trivial rework / code cleanup without any test coverage.

*(or)*

This change is already covered by existing tests, such as *(please describe tests)*.

*(or)*

This change added tests and can be verified as follows:

*(example:)*
  - *Added integration tests for end-to-end deployment with large payloads (10MB)*
  - *Extended integration test for recovery after broker failure*



Gleiphir2769: /pulsarbot run-failure-checks
Gleiphir2769: /pulsarbot run-failure-checks
Gleiphir2769: /pulsarbot run-failure-checks

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..c54ef68 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -469,7 +469,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/producer_partition.go b/pulsar/producer_partition.go
index 881451c..fc564cb 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -480,6 +480,7 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 	var err error
 	if msg.Value != nil && msg.Payload != nil {
 		p.log.Error("Can not set Value and Payload both")
+		request.callback(nil, request.msg, errors.New("can not set Value and Payload both"))
 		return
 	}
 
@@ -493,6 +494,7 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 		if msg.Schema != nil && p.options.Schema != nil &&
 			msg.Schema.GetSchemaInfo().hash() != p.options.Schema.GetSchemaInfo().hash() {
 			p.publishSemaphore.Release()
+			request.callback(nil, request.msg, fmt.Errorf("msg schema can not match with producer schema"))
 			p.log.WithError(err).Errorf("The producer %s of the topic %s is disabled the `MultiSchema`", p.producerName, p.topic)
 			return
 		}
@@ -528,6 +530,7 @@ func (p *partitionProducer) internalSend(request *sendRequest) {
 			if err != nil {
 				p.publishSemaphore.Release()
 				p.log.WithError(err).Error("get schema version fail")
+				request.callback(nil, request.msg, fmt.Errorf("get schema version fail, err: %w", err))
 				return
 			}
 			p.schemaCache.Put(schema.GetSchemaInfo(), schemaVersion)
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..2338074 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"}`
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))
+}