add lock for reassigning partition to new consumer

[yitian108]

When I try to test peer to peer consume using sarama library, some codes snippet like this:

func TestP2P(topic string) {
	if err := NewConsumer(topic, P2PConsumer1); err != nil {
		// some logs
	}

	if err := NewConsumer(topic，P2PConsumer2); err != nil {
		// some logs
	}
	
	stop := false
	count := 0
	go func() {
		for {
			if !stop {
				count++
				msg := "test p2p msg: " + strconv.Itoa(count)
				NewProducer(topic，[]byte(msg))
			}
			time.sleep(time.Second)			
		}
	}()

	select {}
}

Sometimes, running this code will report the panic 'concurrent map writes', and the location of panic points to the method 'addPartitionMovementRecord' in balance_strategy.go, so I add the lock for concurrently running the multiple consumers access the same new partition.

Please check it.

[yitian108]

hello @dnwe, I've modified some codes that only lock one map in the partitionMovements, and the commit of the branch lock-for-adding-partition is 9913274 , hopefully, the newer could be pass all checks.

[dnwe]

@yitian108 are you using NewBalanceStrategySticky for each of your consumers rather than the deprecated shared global of BalanceStrategySticky? They should already have a unique partitionMovements instance if so:

sarama/balance_strategy.go

Lines 107 to 112 in fb81408

	func NewBalanceStrategySticky() BalanceStrategy {
	return &stickyBalanceStrategy{}
	}
	// Deprecated: use NewBalanceStrategySticky to avoid data race issue
	var BalanceStrategySticky = NewBalanceStrategySticky()

[yitian108]

Hi @dnwe , I used the following api for each consumer:

config.Consumer.Group.Rebalance.Strategy = sarama.BalanceStrategySticky

Is this global of BalanceStrategySticky? If that, How should I use the NewBalanceStrategySticky at the starting of config kafka?

ps: I am using the samara version 1.37, which new version using the NewBalanceSTrategySticky?

[dnwe]

It was fix: prevent data race in balance strategy by @napallday from #2453 included in the latest https://github.com/IBM/sarama/releases/tag/v1.40.0

[yitian108]

Thanks @dnwe and @napallday, I'll try to use NewBalanceStrategyRange() instead to test the case.

[yitian108]

Hi @dnwe @napallday, I try to use NewBalanceStrategySticky() as below

	config := sarama.NewConfig()
	config.Consumer.Group.Rebalance.Strategy = sarama.NewBalanceStrategySticky()

However, the IDE GoLand prompts "'Strategy' is deprecated", it seems that I used the wrong way to achive the object, do you have another good way to implement the requirement?

[napallday]

hi @yitian108, it's expected. From Sarama v1.37.0(more precisely PR #2339 ), config.Consumer.Group.Rebalance.Strategy is deprecated and config.Consumer.Group.Rebalance.GroupStrategies is preferred for supporting specifying multiple balance strategies.

[yitian108]

Thanks @napallday , to my understand, I should use like this config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{sarama.NewBalanceStrategySticky(), sarama.NewBalanceStrategyRoundRobin()}, and if only one strategy would be applied, it looks like config.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{sarama.NewBalanceStrategySticky()}, right?

[napallday]

correct! @yitian108

[github-actions]

Thank you for your contribution! However, this pull request has not had any activity in the past 90 days and will be closed in 30 days if no updates occur.
If you believe the changes are still valid then please verify your branch has no conflicts with main and rebase if needed. If you are awaiting a (re-)review then please let us know.