Fixed concurrency issues in SubscriptionManager and BaseDataProducer