diff options
author | Aaron Davidson <aaron@databricks.com> | 2014-11-06 10:45:46 -0800 |
---|---|---|
committer | Aaron Davidson <aaron@databricks.com> | 2014-11-06 10:45:46 -0800 |
commit | 23eaf0e12ff221dcca40a79e61b6cc5e7c846cb5 (patch) | |
tree | f87deebc8e22e13a2f5a74c0e4802a21fa5543d9 /core | |
parent | b41a39e24038876359aeb7ce2bbbb4de2234e5f3 (diff) | |
download | spark-23eaf0e12ff221dcca40a79e61b6cc5e7c846cb5.tar.gz spark-23eaf0e12ff221dcca40a79e61b6cc5e7c846cb5.tar.bz2 spark-23eaf0e12ff221dcca40a79e61b6cc5e7c846cb5.zip |
[SPARK-4264] Completion iterator should only invoke callback once
Author: Aaron Davidson <aaron@databricks.com>
Closes #3128 from aarondav/compiter and squashes the following commits:
698e4be [Aaron Davidson] [SPARK-4264] Completion iterator should only invoke callback once
Diffstat (limited to 'core')
-rw-r--r-- | core/src/main/scala/org/apache/spark/util/CompletionIterator.scala | 5 | ||||
-rw-r--r-- | core/src/test/scala/org/apache/spark/util/CompletionIteratorSuite.scala | 47 |
2 files changed, 51 insertions, 1 deletions
diff --git a/core/src/main/scala/org/apache/spark/util/CompletionIterator.scala b/core/src/main/scala/org/apache/spark/util/CompletionIterator.scala index b6a099825f..390310243e 100644 --- a/core/src/main/scala/org/apache/spark/util/CompletionIterator.scala +++ b/core/src/main/scala/org/apache/spark/util/CompletionIterator.scala @@ -25,10 +25,13 @@ private[spark] // scalastyle:off abstract class CompletionIterator[ +A, +I <: Iterator[A]](sub: I) extends Iterator[A] { // scalastyle:on + + private[this] var completed = false def next() = sub.next() def hasNext = { val r = sub.hasNext - if (!r) { + if (!r && !completed) { + completed = true completion() } r diff --git a/core/src/test/scala/org/apache/spark/util/CompletionIteratorSuite.scala b/core/src/test/scala/org/apache/spark/util/CompletionIteratorSuite.scala new file mode 100644 index 0000000000..3755d43e25 --- /dev/null +++ b/core/src/test/scala/org/apache/spark/util/CompletionIteratorSuite.scala @@ -0,0 +1,47 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.util + +import org.scalatest.FunSuite + +class CompletionIteratorSuite extends FunSuite { + test("basic test") { + var numTimesCompleted = 0 + val iter = List(1, 2, 3).iterator + val completionIter = CompletionIterator[Int, Iterator[Int]](iter, { numTimesCompleted += 1 }) + + assert(completionIter.hasNext) + assert(completionIter.next() === 1) + assert(numTimesCompleted === 0) + + assert(completionIter.hasNext) + assert(completionIter.next() === 2) + assert(numTimesCompleted === 0) + + assert(completionIter.hasNext) + assert(completionIter.next() === 3) + assert(numTimesCompleted === 0) + + assert(!completionIter.hasNext) + assert(numTimesCompleted === 1) + + // SPARK-4264: Calling hasNext should not trigger the completion callback again. + assert(!completionIter.hasNext) + assert(numTimesCompleted === 1) + } +} |