aboutsummaryrefslogtreecommitdiff
path: root/external
diff options
context:
space:
mode:
authorTyson Condie <tcondie@gmail.com>2017-03-23 14:32:05 -0700
committerTathagata Das <tathagata.das1565@gmail.com>2017-03-23 14:32:05 -0700
commit746a558de2136f91f8fe77c6e51256017aa50913 (patch)
tree3325526a325b9774ee1c9a0916210647b1007132 /external
parentb0ae6a38a3ef65e4e853781c5127ba38997a8546 (diff)
downloadspark-746a558de2136f91f8fe77c6e51256017aa50913.tar.gz
spark-746a558de2136f91f8fe77c6e51256017aa50913.tar.bz2
spark-746a558de2136f91f8fe77c6e51256017aa50913.zip
[SPARK-19876][SS][WIP] OneTime Trigger Executor
## What changes were proposed in this pull request? An additional trigger and trigger executor that will execute a single trigger only. One can use this OneTime trigger to have more control over the scheduling of triggers. In addition, this patch requires an optimization to StreamExecution that logs a commit record at the end of successfully processing a batch. This new commit log will be used to determine the next batch (offsets) to process after a restart, instead of using the offset log itself to determine what batch to process next after restart; using the offset log to determine this would process the previously logged batch, always, thus not permitting a OneTime trigger feature. ## How was this patch tested? A number of existing tests have been revised. These tests all assumed that when restarting a stream, the last batch in the offset log is to be re-processed. Given that we now have a commit log that will tell us if that last batch was processed successfully, the results/assumptions of those tests needed to be revised accordingly. In addition, a OneTime trigger test was added to StreamingQuerySuite, which tests: - The semantics of OneTime trigger (i.e., on start, execute a single batch, then stop). - The case when the commit log was not able to successfully log the completion of a batch before restart, which would mean that we should fall back to what's in the offset log. - A OneTime trigger execution that results in an exception being thrown. marmbrus tdas zsxwing Please review http://spark.apache.org/contributing.html before opening a pull request. Author: Tyson Condie <tcondie@gmail.com> Author: Tathagata Das <tathagata.das1565@gmail.com> Closes #17219 from tcondie/stream-commit.
Diffstat (limited to 'external')
-rw-r--r--external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceSuite.scala2
1 files changed, 0 insertions, 2 deletions
diff --git a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceSuite.scala b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceSuite.scala
index 7b6396e029..6391d6269c 100644
--- a/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceSuite.scala
+++ b/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSourceSuite.scala
@@ -301,8 +301,6 @@ class KafkaSourceSuite extends KafkaSourceTest {
StopStream,
StartStream(ProcessingTime(100), clock),
waitUntilBatchProcessed,
- AdvanceManualClock(100),
- waitUntilBatchProcessed,
// smallest now empty, 1 more from middle, 9 more from biggest
CheckAnswer(1, 10, 100, 101, 102, 103, 104, 105, 106, 107,
11, 108, 109, 110, 111, 112, 113, 114, 115, 116,