The subscribedPartitionStates should be thread safe.

The subscribedPartitionState should be thread safe. This variable may be modified by the partition discoverer thread while being read by the main thread. So when subscribed partition changes (e.g. partition expansion, new topic matching the pattern is created), a ConcurrentModificationException may be thrown.

The fix is to use the set from a ConcurrentHashMap instead of a List for subscribedPartitionState.