Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/6707#discussion_r32357897
--- Diff:
streaming/src/test/scala/org/apache/spark/streaming/ReceivedBlockHandlerSuite.scala
---
@@ -174,6 +186,120 @@ class ReceivedBlockHandlerSuite
}
}
+ test("BlockManagerBasedBlockHandler - count messages") {
+ // ByteBufferBlock-MEMORY_ONLY
+ testRecordcount(true, StorageLevel.MEMORY_ONLY,
+ ByteBufferBlock(ByteBuffer.wrap(Array.tabulate(100)(i =>
i.toByte))), blockManager, None)
+ // ArrayBufferBlock-MEMORY_ONLY
+ testRecordcount(true, StorageLevel.MEMORY_ONLY,
+ ArrayBufferBlock(ArrayBuffer.fill(25)(0)), blockManager, Some(25))
+ // ArrayBufferBlock-DISK_ONLY
+ testRecordcount(true, StorageLevel.DISK_ONLY,
+ ArrayBufferBlock(ArrayBuffer.fill(50)(0)), blockManager, Some(50))
+ // ArrayBufferBlock-MEMORY_AND_DISK
+ testRecordcount(true, StorageLevel.MEMORY_AND_DISK,
+ ArrayBufferBlock(ArrayBuffer.fill(75)(0)), blockManager, Some(75))
+ // IteratorBlock-MEMORY_ONLY
+ testRecordcount(true, StorageLevel.MEMORY_ONLY,
+ IteratorBlock((ArrayBuffer.fill(100)(0)).iterator), blockManager,
Some(100))
+ // IteratorBlock-DISK_ONLY
+ testRecordcount(true, StorageLevel.DISK_ONLY,
+ IteratorBlock((ArrayBuffer.fill(125)(0)).iterator), blockManager,
Some(125))
+ // IteratorBlock-MEMORY_AND_DISK
+ testRecordcount(true, StorageLevel.MEMORY_AND_DISK,
+ IteratorBlock((ArrayBuffer.fill(150)(0)).iterator), blockManager,
Some(150))
+ }
+
+ test("WriteAheadLogBasedBlockHandler - count messages") {
+ // ByteBufferBlock-MEMORY_ONLY
+ testRecordcount(false, StorageLevel.MEMORY_ONLY,
+ ByteBufferBlock(ByteBuffer.wrap(Array.tabulate(100)(i =>
i.toByte))), blockManager, None)
+ // ArrayBufferBlock-MEMORY_ONLY
+ testRecordcount(false, StorageLevel.MEMORY_ONLY,
+ ArrayBufferBlock(ArrayBuffer.fill(25)(0)), blockManager, Some(25))
+ // ArrayBufferBlock-DISK_ONLY
+ testRecordcount(false, StorageLevel.DISK_ONLY,
+ ArrayBufferBlock(ArrayBuffer.fill(50)(0)), blockManager, Some(50))
+ // ArrayBufferBlock-MEMORY_AND_DISK
+ testRecordcount(false, StorageLevel.MEMORY_AND_DISK,
+ ArrayBufferBlock(ArrayBuffer.fill(75)(0)), blockManager, Some(75))
+ // IteratorBlock-MEMORY_ONLY
+ testRecordcount(false, StorageLevel.MEMORY_ONLY,
+ IteratorBlock((ArrayBuffer.fill(100)(0)).iterator), blockManager,
Some(100))
+ // IteratorBlock-DISK_ONLY
+ testRecordcount(false, StorageLevel.DISK_ONLY,
+ IteratorBlock((ArrayBuffer.fill(125)(0)).iterator), blockManager,
Some(125))
+ // IteratorBlock-MEMORY_AND_DISK
+ testRecordcount(false, StorageLevel.MEMORY_AND_DISK,
+ IteratorBlock((ArrayBuffer.fill(150)(0)).iterator), blockManager,
Some(150))
+ }
+
+ test("BlockManagerBasedBlockHandler-MEMORY_ONLY - isFullyConsumed") {
+ storageLevel = StorageLevel.MEMORY_ONLY
+ blockManager = createBlockManager(12000)
--- End diff --
This is a pretty subtle leak. The Block Manager created here is going to be
never stopped. Rather than using `createBlockManager` directly, please make it
a `withBlockManager` function (similar to `withBlockManagerBasedBlockHandler`)
that will create the BlockManager, and delete it at the end in a `finally { }`.
So that whenever there is an error, its gets cleaned up without affecting other
tests.
---
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.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
