Repository: apache/pulsar-client-go
Pull request #1034 (2023-06-25 10:43:54)

[fix] Fix ordering key not being set and parsed when batching is disabled

<!--
### 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.
    Skip *Issue XYZ* if there is no associated github issue for this pull request.
    Skip *component* if you are unsure about which is the best component. E.g. `[docs] Fix typo in produce method`.

  - Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
  
  - Each pull request should address only one issue, not mix up code from multiple issues.
  
  - Each commit in the pull request has a meaningful commit message

  - Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.

**(The sections below can be removed for hotfixes of typos)**
-->


### Motivation

When batching is disabled, the ordering key couldn't be configured and parsed correctly. This is a regression bug.
For the reproducible code, please see the test `TestNoBatchSendMessagesWithMetadata` in this PR.

### Modifications

* Set the ordering key when publishing a single message
* Parsed the ordering key when consuming a single message

### Verifying this change


This change added tests: `TestBatchSendMessagesWithMetadata` and `TestNoBatchSendMessagesWithMetadata`

### Does this pull request potentially affect one of the following parts:

*If `yes` was chosen, please highlight the changes*

  - Dependencies (does it add or upgrade a dependency): (yes / no)
  - The public API: (yes / no)
  - The schema: (yes / no / don't know)
  - The default values of configurations: (yes / no)
  - The wire protocol: (yes / no)

### Documentation

  - Does this pull request introduce a new feature? (yes / no)
  - If yes, how is the feature documented? (not applicable / docs / GoDocs / not documented)
  - If a feature is not applicable for documentation, explain why?
  - If a feature is not documented yet in this PR, please create a followup issue for adding the documentation



Patch:
diff --git a/pulsar/client.go b/pulsar/client.go
index 7e6725d..d6f18c9 100644
--- a/pulsar/client.go
+++ b/pulsar/client.go
@@ -120,6 +120,15 @@ type ClientOptions struct {
 	// Configure whether the Pulsar client verify the validity of the host name from broker (default: false)
 	TLSValidateHostname bool
 
+	// TLSCipherSuites is a list of enabled TLS 1.0–1.2 cipher suites. See tls.Config CipherSuites for more information.
+	TLSCipherSuites []uint16
+
+	// TLSMinVersion contains the minimum TLS version that is acceptable. See tls.Config MinVersion for more information.
+	TLSMinVersion uint16
+
+	// TLSMaxVersion contains the maximum TLS version that is acceptable. See tls.Config MaxVersion for more information.
+	TLSMaxVersion uint16
+
 	// Configure the net model for vpc user to connect the pulsar broker
 	ListenerName string
 
diff --git a/pulsar/client_impl.go b/pulsar/client_impl.go
index 9b44987..c283f5a 100644
--- a/pulsar/client_impl.go
+++ b/pulsar/client_impl.go
@@ -91,6 +91,9 @@ func newClient(options ClientOptions) (Client, error) {
 			TrustCertsFilePath:      options.TLSTrustCertsFilePath,
 			ValidateHostname:        options.TLSValidateHostname,
 			ServerName:              url.Hostname(),
+			CipherSuites:            options.TLSCipherSuites,
+			MinVersion:              options.TLSMinVersion,
+			MaxVersion:              options.TLSMaxVersion,
 		}
 	default:
 		return nil, newError(InvalidConfiguration, fmt.Sprintf("Invalid URL scheme '%s'", url.Scheme))
diff --git a/pulsar/consumer_partition.go b/pulsar/consumer_partition.go
index 2d2a194..cd7aa50 100644
--- a/pulsar/consumer_partition.go
+++ b/pulsar/consumer_partition.go
@@ -1190,6 +1190,7 @@ func (pc *partitionConsumer) MessageReceived(response *pb.CommandMessage, header
 				redeliveryCount:     response.GetRedeliveryCount(),
 				schemaVersion:       msgMeta.GetSchemaVersion(),
 				schemaInfoCache:     pc.schemaInfoCache,
+				orderingKey:         string(msgMeta.GetOrderingKey()),
 				index:               messageIndex,
 				brokerPublishTime:   brokerPublishTime,
 			}
diff --git a/pulsar/internal/connection.go b/pulsar/internal/connection.go
index 59aad16..1f80d20 100644
--- a/pulsar/internal/connection.go
+++ b/pulsar/internal/connection.go
@@ -50,6 +50,9 @@ type TLSOptions struct {
 	AllowInsecureConnection bool
 	ValidateHostname        bool
 	ServerName              string
+	CipherSuites            []uint16
+	MinVersion              uint16
+	MaxVersion              uint16
 }
 
 var (
@@ -1046,6 +1049,9 @@ func (c *connection) closed() bool {
 func (c *connection) getTLSConfig() (*tls.Config, error) {
 	tlsConfig := &tls.Config{
 		InsecureSkipVerify: c.tlsOptions.AllowInsecureConnection,
+		CipherSuites:       c.tlsOptions.CipherSuites,
+		MinVersion:         c.tlsOptions.MinVersion,
+		MaxVersion:         c.tlsOptions.MaxVersion,
 	}
 
 	if c.tlsOptions.TrustCertsFilePath != "" {
diff --git a/pulsar/internal/http_client.go b/pulsar/internal/http_client.go
index 2d440d6..e68bd17 100644
--- a/pulsar/internal/http_client.go
+++ b/pulsar/internal/http_client.go
@@ -339,6 +339,9 @@ func getDefaultTransport(tlsConfig *TLSOptions) (http.RoundTripper, error) {
 	if tlsConfig != nil {
 		cfg := &tls.Config{
 			InsecureSkipVerify: tlsConfig.AllowInsecureConnection,
+			CipherSuites:       tlsConfig.CipherSuites,
+			MinVersion:         tlsConfig.MinVersion,
+			MaxVersion:         tlsConfig.MaxVersion,
 		}
 		if len(tlsConfig.TrustCertsFilePath) > 0 {
 			rootCA, err := os.ReadFile(tlsConfig.TrustCertsFilePath)
diff --git a/pulsar/producer_partition.go b/pulsar/producer_partition.go
index 6bd9081..837d1d7 100644
--- a/pulsar/producer_partition.go
+++ b/pulsar/producer_partition.go
@@ -733,6 +733,10 @@ func (p *partitionProducer) genMetadata(msg *ProducerMessage,
 		mm.PartitionKey = proto.String(msg.Key)
 	}
 
+	if len(msg.OrderingKey) != 0 {
+		mm.OrderingKey = []byte(msg.OrderingKey)
+	}
+
 	if msg.Properties != nil {
 		mm.Properties = internal.ConvertFromStringMap(msg.Properties)
 	}
diff --git a/pulsar/single_partition_router.go b/pulsar/single_partition_router.go
new file mode 100644
index 0000000..7896aa3
--- /dev/null
+++ b/pulsar/single_partition_router.go
@@ -0,0 +1,42 @@
+// 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 pulsar
+
+import "sync"
+
+func NewSinglePartitionRouter() func(*ProducerMessage, TopicMetadata) int {
+	var (
+		singlePartition *int
+		once            sync.Once
+	)
+	return func(message *ProducerMessage, metadata TopicMetadata) int {
+		numPartitions := metadata.NumPartitions()
+		if len(message.Key) != 0 {
+			// When a key is specified, use the hash of that key
+			return int(getHashingFunction(JavaStringHash)(message.Key) % numPartitions)
+		}
+		once.Do(func() {
+			partition := r.R.Intn(int(numPartitions))
+			singlePartition = &partition
+		})
+
+		return *singlePartition
+
+	}
+
+}


Test patch:
diff --git a/pulsar/producer_test.go b/pulsar/producer_test.go
index b587975..7c3abdc 100644
--- a/pulsar/producer_test.go
+++ b/pulsar/producer_test.go
@@ -560,6 +560,67 @@ func TestMessageRouter(t *testing.T) {
 	assert.NotNil(t, msg)
 	assert.Equal(t, string(msg.Payload()), "hello")
 }
+func TestMessageSingleRouter(t *testing.T) {
+	// Create topic with 5 partitions
+	topicAdminURL := "admin/v2/persistent/public/default/my-single-partitioned-topic/partitions"
+	err := httpPut(topicAdminURL, 5)
+	defer httpDelete(topicAdminURL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	client, err := NewClient(ClientOptions{
+		URL: serviceURL,
+	})
+
+	assert.Nil(t, err)
+	defer client.Close()
+
+	numOfMessages := 10
+
+	consumer, err := client.Subscribe(ConsumerOptions{
+		Topic:            "my-single-partitioned-topic",
+		SubscriptionName: "my-sub",
+	})
+
+	assert.Nil(t, err)
+	defer consumer.Close()
+
+	producer, err := client.CreateProducer(ProducerOptions{
+		Topic:         "my-single-partitioned-topic",
+		MessageRouter: NewSinglePartitionRouter(),
+	})
+
+	assert.Nil(t, err)
+	defer producer.Close()
+
+	ctx := context.Background()
+
+	for i := 0; i < numOfMessages; i++ {
+		ID, err := producer.Send(ctx, &ProducerMessage{
+			Payload: []byte("hello"),
+		})
+		assert.Nil(t, err)
+		assert.NotNil(t, ID)
+	}
+
+	// Verify message was published on single partition
+	msgCount := 0
+	msgPartitionMap := make(map[string]int)
+	for i := 0; i < numOfMessages; i++ {
+		msg, err := consumer.Receive(ctx)
+		assert.Nil(t, err)
+		assert.NotNil(t, msg)
+		consumer.Ack(msg)
+		msgCount++
+		msgPartitionMap[msg.Topic()]++
+	}
+	assert.Equal(t, msgCount, numOfMessages)
+	assert.Equal(t, len(msgPartitionMap), 1)
+	for _, i := range msgPartitionMap {
+		assert.Equal(t, i, numOfMessages)
+	}
+
+}
 
 func TestNonPersistentTopic(t *testing.T) {
 	topicName := "non-persistent://public/default/testNonPersistentTopic"
@@ -1906,7 +1967,15 @@ func TestMemLimitContextCancel(t *testing.T) {
 	assert.NoError(t, err)
 }
 
-func TestSendMessagesWithMetadata(t *testing.T) {
+func TestBatchSendMessagesWithMetadata(t *testing.T) {
+	testSendMessagesWithMetadata(t, false)
+}
+
+func TestNoBatchSendMessagesWithMetadata(t *testing.T) {
+	testSendMessagesWithMetadata(t, true)
+}
+
+func testSendMessagesWithMetadata(t *testing.T, disableBatch bool) {
 	client, err := NewClient(ClientOptions{
 		URL: lookupURL,
 	})
@@ -1917,7 +1986,7 @@ func TestSendMessagesWithMetadata(t *testing.T) {
 	topic := newTopicName()
 	producer, err := client.CreateProducer(ProducerOptions{
 		Topic:           topic,
-		DisableBatching: true,
+		DisableBatching: disableBatch,
 	})
 	assert.Nil(t, err)
 
@@ -1928,7 +1997,10 @@ func TestSendMessagesWithMetadata(t *testing.T) {
 	assert.Nil(t, err)
 
 	msg := &ProducerMessage{EventTime: time.Now().Local(),
-		Payload: []byte("msg")}
+		Key:         "my-key",
+		OrderingKey: "my-ordering-key",
+		Properties:  map[string]string{"k1": "v1", "k2": "v2"},
+		Payload:     []byte("msg")}
 
 	_, err = producer.Send(context.Background(), msg)
 	assert.Nil(t, err)
@@ -1936,5 +2008,8 @@ func TestSendMessagesWithMetadata(t *testing.T) {
 	recvMsg, err := consumer.Receive(context.Background())
 	assert.Nil(t, err)
 
-	assert.Equal(t, internal.TimestampMillis(recvMsg.EventTime()), internal.TimestampMillis(msg.EventTime))
+	assert.Equal(t, internal.TimestampMillis(msg.EventTime), internal.TimestampMillis(recvMsg.EventTime()))
+	assert.Equal(t, msg.Key, recvMsg.Key())
+	assert.Equal(t, msg.OrderingKey, recvMsg.OrderingKey())
+	assert.Equal(t, msg.Properties, recvMsg.Properties())
 }
diff --git a/pulsar/single_partition_router_test.go b/pulsar/single_partition_router_test.go
new file mode 100644
index 0000000..a5a4215
--- /dev/null
+++ b/pulsar/single_partition_router_test.go
@@ -0,0 +1,62 @@
+// 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 pulsar
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type topicMetaData struct {
+	partition uint32
+}
+
+func (t topicMetaData) NumPartitions() uint32 {
+	return t.partition
+}
+
+func TestNewSinglePartitionRouter(t *testing.T) {
+	numPartitions := topicMetaData{2}
+	router := NewSinglePartitionRouter()
+	p := router(&ProducerMessage{
+		Payload: []byte("message 2"),
+	}, numPartitions)
+	assert.GreaterOrEqual(t, p, 0)
+
+	p2 := router(&ProducerMessage{
+		Payload: []byte("message 2"),
+	}, numPartitions)
+	assert.Equal(t, p, p2)
+}
+
+func TestNewSinglePartitionRouterWithKey(t *testing.T) {
+	router := NewSinglePartitionRouter()
+	numPartitions := topicMetaData{3}
+	p := router(&ProducerMessage{
+		Payload: []byte("message 2"),
+		Key:     "my-key",
+	}, numPartitions)
+	assert.Equal(t, 1, p)
+
+	p2 := router(&ProducerMessage{
+		Key:     "my-key",
+		Payload: []byte("message 2"),
+	}, numPartitions)
+	assert.Equal(t, p, p2)
+}