Github user tdas commented on a diff in the pull request:
https://github.com/apache/spark/pull/5918#discussion_r29900575
--- Diff:
external/flume/src/test/scala/org/apache/spark/streaming/flume/FlumePollingStreamSuite.scala
---
@@ -167,26 +167,24 @@ class FlumePollingStreamSuite extends FunSuite with
BeforeAndAfter with Logging
}
}
- def writeAndVerify(channels: Seq[MemoryChannel], ssc: StreamingContext,
+ def writeAndVerify(sinks: Seq[SparkSink], channels: Seq[MemoryChannel],
ssc: StreamingContext,
outputBuffer: ArrayBuffer[Seq[SparkFlumeEvent]]) {
val clock = ssc.scheduler.clock.asInstanceOf[ManualClock]
val executor = Executors.newCachedThreadPool()
val executorCompletion = new ExecutorCompletionService[Void](executor)
- channels.map(channel => {
+
+ val latch = new CountDownLatch(batchCount * channels.size)
+ sinks.foreach(_.countdownWhenBatchReceived(latch))
+
+ channels.foreach(channel => {
executorCompletion.submit(new TxnSubmitter(channel, clock))
})
+
for (i <- 0 until channels.size) {
executorCompletion.take()
}
- val startTime = System.currentTimeMillis()
- while (outputBuffer.size < batchCount * channels.size &&
- System.currentTimeMillis() - startTime < 15000) {
- logInfo("output.size = " + outputBuffer.size)
- Thread.sleep(100)
- }
- val timeTaken = System.currentTimeMillis() - startTime
- assert(timeTaken < 15000, "Operation timed out after " + timeTaken + "
ms")
- logInfo("Stopping context")
+
+ latch.await(15, TimeUnit.SECONDS)
--- End diff --
Have you seen flakiness in FlumeStream since the last fix that you did ...
with UTF encoding?
And waiting for a 1 second is unreliable. We have to use eventually to wait
for up to 10 seconds.
On Thu, May 7, 2015 at 12:39 PM, Hari Shreedharan <[email protected]>
wrote:
> In
>
external/flume/src/test/scala/org/apache/spark/streaming/flume/FlumePollingStreamSuite.scala
> <https://github.com/apache/spark/pull/5918#discussion_r29888098>:
>
> > for (i <- 0 until channels.size) {
> > executorCompletion.take()
> > }
> > - val startTime = System.currentTimeMillis()
> > - while (outputBuffer.size < batchCount * channels.size &&
> > - System.currentTimeMillis() - startTime < 15000) {
> > - logInfo("output.size = " + outputBuffer.size)
> > - Thread.sleep(100)
> > - }
> > - val timeTaken = System.currentTimeMillis() - startTime
> > - assert(timeTaken < 15000, "Operation timed out after " + timeTaken
+ " ms")
> > - logInfo("Stopping context")
> > +
> > + latch.await(15, TimeUnit.SECONDS)
>
> Actually, I think it is simpler to do latch.await + manual clock that
> moves couple of batch intervals forward - so we know data is sent + data
is
> processed and wait for 1 second for processing to complete. We have seen
> some flakiness even with the FlumeStream approach.
>
> â
> Reply to this email directly or view it on GitHub
> <https://github.com/apache/spark/pull/5918/files#r29888098>.
>
---
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]
