[Issue 1027][producer] fix: split sendRequest and make reconnectToBroker and other operate in the same coroutine

[graysonzeng]

Fixes #1027

Motivation

fix send timeout error cause by reconnection failures

Modifications

After #691 , client avoid blocking the reconnection logic, but it also make reconnection and internalSend/internalFlush have concurrency issues。

when the client connection is block, it run into p.grabCnx() to get new connection and resend pendingQueue messages. At the same time, the internalSend/internalFlush keep running by use pendingQueue and send mssage by p._getConn().WriteData. Therefore, it received an ack larger than expected, and trigger the ConnectionClosed logic, then the client enter p.reconnectToBroker() and repeat the concurrency issues with internalSend/internalFlush, fall into a reconnect-resend-connectionClosed loop

Verifying this change

• Make sure that the change passes the CI checks.

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: (no)
• The schema: (no)
• The default values of configurations: (no)
• The wire protocol: (no)

Documentation

• Does this pull request introduce a new feature? (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

[graysonzeng]

BTW, I think @bschofield about some guess said in #697 is right.

[shibd]

@zengguan Thanks for your contribute.

I have a question, As #687. Can we let the sendRequest event alone in one channel?

In this way, Let reconnectToBroker and other operate are kept in the same coroutine.

	for {
		select {
		case sendRequest <-p.sendRequestChan:
		    p.internalSend(sendRequest)
		case i := <-p.eventsChan:
			switch v := i.(type) {
			case *flushRequest:
				p.internalFlush(v)
			case *closeProducer:
				p.internalClose(v)
				return
			}
		case <-p.connectClosedCh:
			p.reconnectToBroker()
		case <-p.batchFlushTicker.C:
			if p.batchBuilder.IsMultiBatches() {
				p.internalFlushCurrentBatches()
			} else {
				p.internalFlushCurrentBatch()
			}
		}
	}

I think we can discuss clearly here, using channels should avoid using locks.

[graysonzeng]

@shibd great idea ! Now I split sendRequest and changed eventsChan to dataChan and cmdChan to handle sendRequest and flushRequest/closeProducer separately。I've tested it and it doesn't reproduce anymore

[shibd]

@BewareMyPower @gaoran10 Can you help take a look?

[BewareMyPower]

Could you update your PR description since you don't use RWLock now?

[BewareMyPower]

It seems the CI is broken.

[graysonzeng]

Could you update your PR description since you don't use RWLock now?

Done.

[BewareMyPower]

Please rebase to master to have tests fixed.