KAFKA-5147: Add missing synchronization to TransactionManager #3132
Conversation
|
cc @hachikuji |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
|
@apurvam Please rebase when you get a chance. Thanks! |
…ds) and correcness (for all accesses to the currentState)
apurvam
force-pushed
the
KAFKA-5147-transaction-manager-synchronization-fixes
branch
from
55b7197
to
3f3b036
Compare
|
@hachikuji I rebased onto trunk and added more synchronization for both consistency and correctness as we discussed today. |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
| @@ -769,7 +769,7 @@ boolean isEndTxn() { | |||
| } | |||
|
|
|||
| @Override | |||
| public void handleResponse(AbstractResponse response) { | |||
| public synchronized void handleResponse(AbstractResponse response) { | |||
There was a problem hiding this comment.
This and the other handlers should synchronize on TransactionManager.this, right?
There was a problem hiding this comment.
You are right. Fixed.
|
One other thing I noticed. The |
|
@hachikuji I addressed your latest comments. Please take a look. |
| @@ -534,7 +534,9 @@ public void onComplete(ClientResponse response) { | |||
| fatalError(response.versionMismatch()); | |||
| } else if (response.hasResponse()) { | |||
| log.trace("Got transactional response for request:" + requestBuilder()); | |||
| handleResponse(response.responseBody()); | |||
| synchronized (TransactionManager.this) { | |||
| @@ -515,7 +515,7 @@ void fail(RuntimeException e) { | |||
| result.done(); | |||
| } | |||
|
|
|||
| void reenqueue() { | |||
| synchronized void reenqueue() { | |||
There was a problem hiding this comment.
This one too should synchronize on TransactionManager.this.
There was a problem hiding this comment.
Ack. Thanks for pointing these out. I am learning things!
|
I'll fix the checkstyle error when I merge. |
|
Refer to this link for build results (access rights to CI server needed): |
|
Refer to this link for build results (access rights to CI server needed): |
The basic idea is that exactly three collections, ie. `pendingRequests`, `newPartitionsToBeAddedToTransaction`, and `partitionsInTransaction` are accessed from the context of application threads. The first two are modified from the application threads, and the last is read from those threads. So to make the `TransactionManager` truly thread safe, we have to ensure that all accesses to these three members are done in a synchronized block. I inspected the code, and I believe this patch puts the synchronization in all the correct places. Author: Apurva Mehta <apurva@confluent.io> Reviewers: Jason Gustafson <jason@confluent.io> Closes #3132 from apurvam/KAFKA-5147-transaction-manager-synchronization-fixes (cherry picked from commit 02c0c3b) Signed-off-by: Jason Gustafson <jason@confluent.io>
The basic idea is that exactly three collections, ie.
pendingRequests,newPartitionsToBeAddedToTransaction, andpartitionsInTransactionare accessed from the context of application threads. The first two are modified from the application threads, and the last is read from those threads.So to make the
TransactionManagertruly thread safe, we have to ensure that all accesses to these three members are done in a synchronized block. I inspected the code, and I believe this patch puts the synchronization in all the correct places.