Repository: apache/pulsar-client-go
Pull request #999 (2023-04-01 03:38:18)

Add single partition router
### Motivation
![image](https://user-images.githubusercontent.com/15684156/229263528-dece94ed-0df7-4159-9c60-5489a4f7da9f.png)

https://pulsar.apache.org/docs/2.11.x/concepts-messaging/#routing-modes
https://github.com/apache/pulsar/blob/master/pulsar-client/src/main/java/org/apache/pulsar/client/impl/SinglePartitionMessageRouterImpl.java

Based on the code of the Java client, Implemented the `single-partition-router` mentioned in the document.


### Modifications

Add single-partition-router

### Verifying this change

- [ ] Make sure that the change passes the CI checks.


This change added tests and can be verified as follows:

- https://github.com/crossoverJie/pulsar-client-go/blob/add-single-partition-router/pulsar/single_partition_router_test.go
- https://github.com/crossoverJie/pulsar-client-go/blob/add-single-partition-router/pulsar/single_partition_route_bench_test.go
- https://github.com/crossoverJie/pulsar-client-go/blob/ae50c8bd6a960410069e2568ea9aeb1e4d27a625/pulsar/producer_test.go#L563



### 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): (no)
  - The public API: (yes) Add `NewSinglePartitionRouter()` method.
  - The schema: (no)
  - The default values of configurations: (no)
  - The wire protocol: (no)

### Documentation

  - Does this pull request introduce a new feature? (yes)
  - If yes, how is the feature documented? (GoDocs)


shibd: @crossoverJie Hi, thanks for your contribution. Can you fix the ci lint failed?
crossoverJie: > @crossoverJie Hi, thanks for your contribution. Can you fix the ci lint failed?
PTAL
crossoverJie: /pulsarbot run-failure-checks
BewareMyPower: Please rebase to master to have tests fixed.
crossoverJie: > Please rebase to master to have tests fixed.

Done
crossoverJie: > Thank you for your contribution! But the default router supports this feature:
> 
> https://github.com/apache/pulsar-client-go/blob/a119bab0f8598601c0eb7f0fcd97da7ab06700c7/pulsar/default_router.go#L57-L65

But when there are no `OrderdingKey` and `Key`, the user needs to specify specific partitions manually, here refers to the API of the Java SDK, where the SDK selects the partitions, keeping consistent with the implementation of the Java SDK.

https://github.com/apache/pulsar/blob/46b6dcd9571262f72f963a755c8a51fb4f66c81e/pulsar-broker/src/test/java/org/apache/pulsar/client/api/SimpleProducerConsumerStatTest.java#L153C1-L155
crossoverJie: CI run failed, I created another [PR](https://github.com/apache/pulsar-client-go/pull/1037) to fix this issue.



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/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/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..2721fa3 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"
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)
+}