Add single partition router

[crossoverJie]

Motivation

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

[shibd]

We need to judge again here if singlePartition is nil. Set it value only when it is nil.

[crossoverJie]

Would using sync.Once be better?

[shibd]

Agree.

[crossoverJie]

Done

Misoperation approved

[nodece]

Thank you for your contribution! But the default router supports this feature:

pulsar-client-go/pulsar/default_router.go

Lines 57 to 65 in a119bab

	if len(message.OrderingKey) != 0 {
	// When an OrderingKey is specified, use the hash of that key
	return int(hashFunc(message.OrderingKey) % numPartitions)
	}
	if len(message.Key) != 0 {
	// When a key is specified, use the hash of that key
	return int(hashFunc(message.Key) % numPartitions)
	}

[crossoverJie]

Thank you for your contribution! But the default router supports this feature:

pulsar-client-go/pulsar/default_router.go

Lines 57 to 65 in a119bab

	if len(message.OrderingKey) != 0 {
	// When an OrderingKey is specified, use the hash of that key
	return int(hashFunc(message.OrderingKey) % numPartitions)
	}
	if len(message.Key) != 0 {
	// When a key is specified, use the hash of that key
	return int(hashFunc(message.Key) % numPartitions)
	}

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

[nodece]

DO NOT use the global variable.

Please move these variables to line 28.

[nodece]

I notice that Java only checks the message.key, without message.orderingKey, so we should keep the same implementation.

[crossoverJie]

CI run failed, I created another PR to fix this issue.

[shibd]

@nodece Hi, Can you take a look at it again?