Fix transaction coordinator client cann't reconnect to the broker

[RobertIndie]

Fixes #1227

Motivation

There are some issues with the transactionCoordinatorClient in the go client. When using the transaction, if there are any reconnection happens during the transaction operation. The connection to the transaction coordinator won't be reconnected. This causes all following operations to fail with the connection closed error.

Modifications

• Introduced transactionHandler to manage reconnections and handle requests.
• Fix the tc client would crash if the broker doesn't enable the transaction
• Improved the error handling in the tc client.

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

[BewareMyPower]

This PR might be over complicated.

IIUC, the root cause is that when RequestOnCnx returns an error that indicates the connection is closed, it just propagate to the caller side. Here is a simple fix (without any backoff mechanism and timeout):

func (tc *transactionCoordinatorClient) sendRequest(cnxIndex uint64, requestID uint64, cmdType pb.BaseCommand_Type, message proto.Message) error {
        for {
                res, err := tc.client.rpcClient.RequestOnCnx(tc.cons[cnxIndex], requestID, cmdType, message)
                if err != nil {
                        if err.Error() != "connection closed" {
                                return err
                        }
                        if err = tc.grabConn(cnxIndex); err != nil {
                                return err
                        }
                        continue
                }
                if res.Response.EndTxnResponse.Error != nil {
                        return getErrorFromServerError(res.Response.EndTxnResponse.Error)
                }
                return nil
        }
}

Then replace the RequestOnCnx call like:

        err := tc.sendRequest(id.MostSigBits, requestID, pb.BaseCommand_END_TXN, cmdEndTxn)
        tc.semaphore.Release()
        if err != nil {
                return err
        }
        return nil

The fix above will bring much less code changes and the code will look more simple.

[RobertIndie]

The fix above will bring much less code changes and the code will look more simple.

This implementation misses some key details. For example, it lacks a backoff mechanism and connection timeout logic as you said. Also, what happens if many goroutines call grabConn at the same time? We might end up with multiple goroutines trying to handle reconnections simultaneously. This doesn't seem like a good approach. Although we could use locks or other methods to manage concurrent executions, there are still issues. What if there's a connection timeout and we need to notify the user? Plus, how do we inform other goroutines to stop attempting reconnections in that case? It seems we need complex and less intuitive logic to address these issues.

IIUC, the main simplification in this implementation involves transfering the XXOp and the event loop logic, right? The key difference appears to be the use of the event loop model. However, reducing this complexity doesn't seem very significant to me. Both producers and consumers already use the event loop for handling requests.

pulsar-client-go/pulsar/consumer_partition.go

Lines 1578 to 1582 in b0111a2

	func (pc *partitionConsumer) runEventsLoop() {
	defer func() {
	pc.log.Debug("exiting events loop")
	}()
	pc.log.Debug("get into runEventsLoop")

The Java implementation of the TransactionMetaStoreHandler also uses something like event loop, similar to Go:
TransactionMetaStoreHandler Java code

Moreover, without the event loop, it seems challenging to implement asynchronous methods like XXXAsync. For example, Producer.SendAsync. Although we don't have this feature yet, it would be beneficial to make these operations truly asynchronous in the future. The Java client already supports this feature:
Async support in Java client

This PR mainly adopts the eventloop patterns of the the existing producer and consumer. Maybe it's better to start a separate discussion for this.

[BewareMyPower]

I see, currently the Context object is also not used so we need an event loop to handle it.