Github user harishreedharan commented on a diff in the pull request:
https://github.com/apache/spark/pull/1195#discussion_r14305990
--- Diff:
streaming/src/main/scala/org/apache/spark/streaming/receiver/BlockGenerator.scala
---
@@ -58,6 +66,11 @@ private[streaming] class BlockGenerator(
@volatile private var currentBuffer = new ArrayBuffer[Any]
@volatile private var stopped = false
+ private var currentBlockId: StreamBlockId = StreamBlockId(receiverId,
+ clock.currentTime() - blockInterval)
+
+ // Removes might happen from the map while other threads are inserting.
--- End diff --
Not really. We have to protect against other threads having a reference to
the ArrayBuffer corresponding to each block id. Specifically if a thread is in
the store method and is adding values to the buffer, and another thread calls
remove() on the same block id from the map - the buffer could still be changing
while the 2nd thread is calling the callbacks. To prevent this, any operation
on the buffer and removal from the map should be protected by the same lock. So
any += calls to the buffer and any removes from the map should be synchronized.
This ensures that there is no thread holding onto a reference of the buffer
instance while the buffer is being removed from the map.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
