Skip to content

[SPARK-8404][Streaming][Tests] Use thread-safe collections to make the tests more reliable #6852

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
wants to merge 1 commit into from
Closed

[SPARK-8404][Streaming][Tests] Use thread-safe collections to make the tests more reliable #6852

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 2 additions & 4 deletions ...rnal/kafka/src/test/java/org/apache/spark/streaming/kafka/JavaDirectKafkaStreamSuite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,7 @@
package org.apache.spark.streaming.kafka;

import java.io.Serializable;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Arrays;
import java.util.*;

import scala.Tuple2;

Expand Down Expand Up @@ -116,7 +114,7 @@ public String call(MessageAndMetadata<String, String> msgAndMd) throws Exception
);
JavaDStream<String> unifiedStream = stream1.union(stream2);

final HashSet<String> result = new HashSet<String>();
final Set<String> result = Collections.synchronizedSet(new HashSet<String>());
Copy link
Member

@srowen srowen Jun 17, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When would this be run concurrently though?

unifiedStream.foreachRDD(
new Function<JavaRDD<String>, Void>() {
@Override
Expand Down
6 changes: 2 additions & 4 deletions external/kafka/src/test/java/org/apache/spark/streaming/kafka/JavaKafkaStreamSuite.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,9 +18,7 @@
package org.apache.spark.streaming.kafka;

import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Random;
import java.util.*;

import scala.Tuple2;

Expand Down Expand Up @@ -94,7 +92,7 @@ public void testKafkaStream() throws InterruptedException {
topics,
StorageLevel.MEMORY_ONLY_SER());

final HashMap<String, Long> result = new HashMap<String, Long>();
final Map<String, Long> result = Collections.synchronizedMap(new HashMap<String, Long>());

JavaDStream<String> words = stream.map(
new Function<Tuple2<String, String>, String>() {
Expand Down
14 changes: 8 additions & 6 deletions external/kafka/src/test/scala/org/apache/spark/streaming/kafka/DirectKafkaStreamSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,7 +99,8 @@ class DirectKafkaStreamSuite
ssc, kafkaParams, topics)
}

val allReceived = new ArrayBuffer[(String, String)]
val allReceived =
new ArrayBuffer[(String, String)] with mutable.SynchronizedBuffer[(String, String)]

stream.foreachRDD { rdd =>
// Get the offset ranges in the RDD
Expand Down Expand Up @@ -162,7 +163,7 @@ class DirectKafkaStreamSuite
"Start offset not from latest"
)

val collectedData = new mutable.ArrayBuffer[String]()
val collectedData = new mutable.ArrayBuffer[String]() with mutable.SynchronizedBuffer[String]
stream.map { _._2 }.foreachRDD { rdd => collectedData ++= rdd.collect() }
ssc.start()
val newData = Map("b" -> 10)
Expand Down Expand Up @@ -208,7 +209,7 @@ class DirectKafkaStreamSuite
"Start offset not from latest"
)

val collectedData = new mutable.ArrayBuffer[String]()
val collectedData = new mutable.ArrayBuffer[String]() with mutable.SynchronizedBuffer[String]
stream.foreachRDD { rdd => collectedData ++= rdd.collect() }
ssc.start()
val newData = Map("b" -> 10)
Expand Down Expand Up @@ -324,7 +325,8 @@ class DirectKafkaStreamSuite
ssc, kafkaParams, Set(topic))
}

val allReceived = new ArrayBuffer[(String, String)]
val allReceived =
new ArrayBuffer[(String, String)] with mutable.SynchronizedBuffer[(String, String)]

stream.foreachRDD { rdd => allReceived ++= rdd.collect() }
ssc.start()
Expand All @@ -350,8 +352,8 @@ class DirectKafkaStreamSuite
}

object DirectKafkaStreamSuite {
val collectedData = new mutable.ArrayBuffer[String]()
var total = -1L
val collectedData = new mutable.ArrayBuffer[String]() with mutable.SynchronizedBuffer[String]
@volatile var total = -1L

class InputInfoCollector extends StreamingListener {
val numRecordsSubmitted = new AtomicLong(0L)
Expand Down
7 changes: 2 additions & 5 deletions external/kafka/src/test/scala/org/apache/spark/streaming/kafka/KafkaStreamSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,7 @@ class KafkaStreamSuite extends SparkFunSuite with Eventually with BeforeAndAfter

val stream = KafkaUtils.createStream[String, String, StringDecoder, StringDecoder](
ssc, kafkaParams, Map(topic -> 1), StorageLevel.MEMORY_ONLY)
val result = new mutable.HashMap[String, Long]()
val result = new mutable.HashMap[String, Long]() with mutable.SynchronizedMap[String, Long]
stream.map(_._2).countByValue().foreachRDD { r =>
val ret = r.collect()
ret.toMap.foreach { kv =>
Expand All @@ -77,10 +77,7 @@ class KafkaStreamSuite extends SparkFunSuite with Eventually with BeforeAndAfter
ssc.start()

eventually(timeout(10000 milliseconds), interval(100 milliseconds)) {
assert(sent.size === result.size)
sent.keys.foreach { k =>
assert(sent(k) === result(k).toInt)
}
assert(sent === result)
}
}
}