Fix race condition/goroutine leak in partition discovery goroutine

[PowerStateFailure]

Master issue: #462

Motivation

In some cases consumer Close method might be called in nearly the same moment with background partiotion discovery routine tick. When Close acquires mutex first, it will release all underlying consumers, but that not guarantees backgound routine's not wait for the same mutex in internalTopicSubscribeToPartitions, so when Close completes, mutex gets acquired by not-finished-yet goroutine leading to new partition consumers gets created without any chance to close them.
The same logic is valid for producer partition discovery goroutine which is closed the same manner.

This PR modifies background discrovery goroutine the way it will be closed before acquiring mutex in Close method.

Modifications

Described above.

Verifying this change

• Make sure that the change passes the CI checks.

This change is a trivial rework / code cleanup without any test coverage.

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

[omnilight]

@PowerStateFailure I faced the same problem. My service is a proxy between foreign clients and pulsar. When foreign client connects/disconnects too often and we have changed number of partitions for topic, they could get another consumer for partition.

Could you please fix failed build?

@wolfstudy Could you please review this PR?

[PowerStateFailure]

@omnilight

Could you please fix failed build?

As i can see from build log, there is race condition somewhere in connection acquire logic. As I don't change it, i'm unsure if i should fix it.

BTW, it seems that producer has the same issue with closing background partition discovery goroutine, so I'll try to fix it too.

[PowerStateFailure]

Rebased to the latest master, thus avoiding undesirable data race error in tests.

[merlimat]

👍