[SPARK-3686][STREAMING] Wait for sink to commit the channel before ch…

…ecking for the channel size.
apache · Sep 25, 2014 · 6ce9d8b · 6ce9d8b
1 parent 74fb2ec
commit 6ce9d8b
Showing 1 changed file with 2 additions and 0 deletions.
diff --git a/...rnal/flume-sink/src/test/scala/org/apache/spark/streaming/flume/sink/SparkSinkSuite.scala b/...rnal/flume-sink/src/test/scala/org/apache/spark/streaming/flume/sink/SparkSinkSuite.scala
@@ -51,6 +51,7 @@ class SparkSinkSuite extends FunSuite {
     val events = client.getEventBatch(1000)
     client.ack(events.getSequenceNumber)
     assert(events.getEvents.size() === 1000)
+    TimeUnit.SECONDS.sleep(1) // Allow the sink to commit the transactions.
     assertChannelIsEmpty(channel)
     sink.stop()
     channel.stop()
@@ -70,6 +71,7 @@ class SparkSinkSuite extends FunSuite {
     val events = client.getEventBatch(1000)
     assert(events.getEvents.size() === 1000)
     client.nack(events.getSequenceNumber)
+    TimeUnit.SECONDS.sleep(1) // Allow the sink to commit the transactions.
     assert(availableChannelSlots(channel) === 4000)
     sink.stop()
     channel.stop()