[Transaction] Add transaction coordinator disconnect command.
- Dominant language
- Java
- Stars
- 15.3k
- Forks
- 3.8k
- Avg merge
- 1d 14h
- Merged PRs (30d)
- 160
Description
# Background
Now, transaction coordinator client connect coordinator use `CommandTcClientConnectRequest` , when CoordinatorClient connect to coordinator will lookup `transaction_coordinator_assign` topic and then find coordinator broker then send `CommandTcClientConnectRequest` to broker, broker will init tc.
Tc client disconnect from coordinator when do transaction op throw exception `TransactionCoordinatorNotFound` , this will cause tc connect don't be cleaned up.
# Motivation
## Clean up tc connect
We need to clean up tc connect to prevent some unnecessary connections and occupy resources
## Release transaction resources
Now, when tc client disconnect, the only way to end a transaction is to use transaction timeout. It will have a lot of bad influences eg.
1. If you ack messages with transaction and then then this handler is closed additional consumption program will receive these messages and ack with transaction. In this time, the new handler will receive exception `TransactionConflictException` , because the message has in pending Ack state and the transaction have not been ended. When the pervious transaction end by timeout, these ack messages will back to normal. Therefore, it greatly affects the continuity of consumption and will always wait for the previous transaction timeout under cumulative ack.
2. As we know, pulsar transaction currently only supports streaming transactions so we have the concept of a StablePosition. If the transaction don't end in time, It will affect the movement of stableposition, resulting in consumers not being able to receive unaffected messages in time
So, we should provide the config of when tc client disconnect, we can abort all of the ongoing transaction created by this tc client connect.
# implement
## Change proto
### CommandTcClientDisconnectRequest
```protobuf
message CommandTcClientDisconnectRequest {
required string connect_name = 1;
required uint64 request_id = 2;
}
```
The command is used to disconnect tc client connect from broker to client or from client to broker.
Now, the connect name generate by transaction coordinator and with prefix `Automatically_Generated_` + AtomicLong.incrementAndGet().
Because I can’t think of what the clientname does for the time being, we turn on the client-specified generation function, but we reserve the clientName whose String is not long to prevent the need for the user to specify the generated clientName in the future.
### CommandTcClientConnectRequest
```protobuf
message CommandTcClientConnectRequest {
required uint64 request_id = 1;
required uint64 tc_id = 2 [default = 0];
//Add field
optional bool close_connect_abort_txn = 3 [ default = false ];
}
```
When client send tc client connect command to coordinator will carry field `close_connect_abort_txn` then coordinator will know the transaction created by this client needs to be aborted after disconnection.
#####Tip: Should we need to think of the connect will created two way transaction? Disconnect abort or disconnect don't abort.
###CommandTcClientConnectResponse
```protobuf
message CommandTcClientConnectResponse {
required uint64 request_id = 1;
optional ServerError error = 2;
optional string message = 3;
optional uint64 tc_client_id = 4;
}
```
Add `optional string connect_name = 4;`, when tc client actively close the connection will use client-name so client should know the connect_name.
### TransactionMetadataEntry
```protobuf
message TransactionMetadataEntry {
enum TransactionMetadataOp {
NEW = 0;
ADD_PARTITION = 1;
ADD_SUBSCRIPTION = 2;
UPDATE = 3;
}
optional TransactionMetadataOp metadata_op = 1;
optional uint64 txnid_least_bits = 2 [default = 0];
optional uint64 txnid_most_bits = 3 [default = 0];
optional TxnStatus expected_status = 4;
optional TxnStatus new_status = 5;
repeated string partitions = 6;
repeated pulsar.proto.Subscription subscriptions = 7;
optional uint64 timeout_ms = 8;
optional uint64 start_time = 9;
optional uint64 last_modification_time = 10;
optional uint64 max_local_txn_id = 11;
optional bool close_connect_abort_txn = 3 [ default = false ];
}
```
Because when Tc coordinator recover, we need to know which transaction needs to be aborted, because the client that generated it specifies that it needs to be aborted after disconnecting
### Code change continually updated
discuss record :
2021.12.20
1. recover and reconnect should abort transaction?
2. disconnect should abort transaction, but does it really need to abort?
3. Do we need to restrict the transaction can only be operated with the client that generated it?
Contributor guide
Assessment
This issue has not been assessed yet.