KAFKA-7284: streams should unwrap fenced exception #5520
Conversation
|
@mjsax @guozhangwang , Here's the 1.0 implementation. The RecordCollectorImpl is significantly different from the later branches. I tried to preserve the existing logic, but it might we worth a close look. Thanks, |
| @@ -205,7 +206,7 @@ public void abortTransaction() throws ProducerFencedException { | |||
| this.transactionInFlight = false; | |||
| } | |||
|
|
|||
| private void verifyProducerState() { | |||
| private synchronized void verifyProducerState() { | |||
There was a problem hiding this comment.
why do we need to synchronized a private method?
I was just looking into trunk and it seems to be synchronized there, too. Also, some methods are synchronized others are not, but I could not see any pattern. Should we just synchronized all public method? Performance should not be a concern for MockProducer. On the other hand, when would a producer be used by multiple threads? Neither KafkaConsumer nor KafkaProducer are thread-safe by design. Why does MockProducer need to be thread-safe?
There was a problem hiding this comment.
I added it in the prior PRs because findbugs complained that some, but not all, accesses of the "am I fenced" field are synchronized. I didn't bother verifying that it's necessary in the older branches, but I can check it out real quick.
There was a problem hiding this comment.
I don't know why anything in here is synchronized; that's just the way I found it.
There was a problem hiding this comment.
I see. Maybe ok for older branches to leave as is. But might be wroth to cleanup in trunk?
There was a problem hiding this comment.
To be fair: fenceProducer() is not part of public producer API but added to MockProducer and it might be called from a different thread (ie, the main testing thread while StreamThread used the producer, too). Because all public methods within MockProducer call verifyState() it makes sense what findbugs complains about. Did miss this connection before. So maybe drop latest commit :)
Sorry for the forth and back.
There was a problem hiding this comment.
Turns out the tests fail with the synchronization is removed. See https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/15966/findbugsResult/ while it's still available.
|
I missed the failure messages. Retest this, please. |
|
@mjsax The test failure was due to the missing synchronization, so I've added it back in. |
|
Retest this please |
1 similar comment
|
Retest this please |
|
Note that the java 10 build won't pass on 1.x due to the incompatible scala version. |
|
Merged to |
Reviewers: Matthias J. Sax <matthias@confluent.io>
Unwrap the ProducerFencedException in RecordCollectorImpl
so it can be caught and converted to a TaskMigratedException
instead of triggering a shutdown.
See also #5499 and #5513
Committer Checklist (excluded from commit message)