KAFKA-9668: Iterating over KafkaStreams.getAllMetadata() results in ConcurrentModificationException #8233
Conversation
…oncurrentModificationException `KafkaStreams.getAllMetadata()` returns `StreamsMetadataState.getAllMetadata()`. All the latter methods is `synchronized` it returns a reference to internal mutable state. Not only does this break encapsulation, but it means any thread iterating over the returned collection when the metadata gets rebuilt will encounter a `ConcurrentModificationException`. This change: * switches from clearing and rebuild `allMetadata` when `onChange` is called to building a new list and swapping this in. This is thread safe and has the benefit that the returned list is not empty during a rebuild: you either get the old or the new list. * removes synchronisation from `getAllMetadata` and `getLocalMetadata`. These are returning member variables. Synchronisation adds nothing. * changes `getAllMetadata` to wrap its return value in an unmodifiable wrapper to avoid breaking encapsulation. * changes the getters in `StreamsMetadata` to wrap their return values in unmodifiable wrapper to avoid breaking encapsulation.
|
This has obviously been a long standing bug. So no specific urgency to get it in. It is causing build failures in our build, but I'm going to try and work around it. |
| if (activePartitionHostMap.isEmpty() && standbyPartitionHostMap.isEmpty()) { | ||
| allMetadata = Collections.emptyList(); |
There was a problem hiding this comment.
Why move the reset into the condition? if the passed in values are empty we should still set it to empty because the previous map may be out-dated right?
There was a problem hiding this comment.
Well, with this change you're always going to get either the old or new metadata from getAllMetadata. Where as if I clear the list at the start of this method, then getAllMetadata can return empty.
Question is: which is preferable? I figured there's a strong possibility that at least some of the old metadata may still be correct - so why ditch it?
There was a problem hiding this comment.
By returning empty we're forcing callers to handle this case. But if we return stale data, then it's no different to if they had called getAllMetadata a few ns earlier.
|
test this please |
|
test this please |
|
retest this please |
1 similar comment
|
retest this please |
|
LGTM. Merging to trunk and 2.5 now. |
…oncurrentModificationException (#8233) `KafkaStreams.getAllMetadata()` returns `StreamsMetadataState.getAllMetadata()`. All the latter methods is `synchronized` it returns a reference to internal mutable state. Not only does this break encapsulation, but it means any thread iterating over the returned collection when the metadata gets rebuilt will encounter a `ConcurrentModificationException`. This change: * switches from clearing and rebuild `allMetadata` when `onChange` is called to building a new list and swapping this in. This is thread safe and has the benefit that the returned list is not empty during a rebuild: you either get the old or the new list. * removes synchronisation from `getAllMetadata` and `getLocalMetadata`. These are returning member variables. Synchronisation adds nothing. * changes `getAllMetadata` to wrap its return value in an unmodifiable wrapper to avoid breaking encapsulation. * changes the getters in `StreamsMetadata` to wrap their return values in unmodifiable wrapper to avoid breaking encapsulation. Co-authored-by: Andy Coates <[email protected]> Reviewers: Guozhang Wang <[email protected]>
Fixes KAFKA-9668
KafkaStreams.getAllMetadata()returnsStreamsMetadataState.getAllMetadata(). All the latter methods issynchronizedit returns a reference to internal mutable state. Not only does this break encapsulation, but it means any thread iterating over the returned collection when the metadata gets rebuilt will encounter aConcurrentModificationException.This change:
allMetadatawhenonChangeis called to building a new list and swapping this in. This is thread safe and has the benefit that the returned list is not empty during a rebuild: you either get the old or the new list.getAllMetadataandgetLocalMetadata. These are returning member variables. Synchronisation adds nothing.getAllMetadatato wrap its return value in an unmodifiable wrapper to avoid breaking encapsulation.StreamsMetadatato wrap their return values in unmodifiable wrapper to avoid breaking encapsulation.Unit tests have been added to cover both changes classes to ensure encapsulation and thread-safety are maintained.
Committer Checklist (excluded from commit message)