diff options
| author | Shixiong Zhu <shixiong@databricks.com> | 2016-02-02 21:13:54 -0800 |
|---|---|---|
| committer | Reynold Xin <rxin@databricks.com> | 2016-02-02 21:13:54 -0800 |
| commit | 335f10edad8c759bad3dbd0660ed4dd5d70ddd8b (patch) | |
| tree | 997aac293bb3f2f7bc565693e5800f13ab31db2c /python/pyspark/streaming/tests.py | |
| parent | 0557146619868002e2f7ec3c121c30bbecc918fc (diff) | |
| download | spark-335f10edad8c759bad3dbd0660ed4dd5d70ddd8b.tar.gz spark-335f10edad8c759bad3dbd0660ed4dd5d70ddd8b.tar.bz2 spark-335f10edad8c759bad3dbd0660ed4dd5d70ddd8b.zip | |
[SPARK-7997][CORE] Add rpcEnv.awaitTermination() back to SparkEnv
`rpcEnv.awaitTermination()` was not added in #10854 because some Streaming Python tests hung forever.
This patch fixed the hung issue and added rpcEnv.awaitTermination() back to SparkEnv.
Previously, Streaming Kafka Python tests shutdowns the zookeeper server before stopping StreamingContext. Then when stopping StreamingContext, KafkaReceiver may be hung due to https://issues.apache.org/jira/browse/KAFKA-601, hence, some thread of RpcEnv's Dispatcher cannot exit and rpcEnv.awaitTermination is hung.The patch just changed the shutdown order to fix it.
Author: Shixiong Zhu <shixiong@databricks.com>
Closes #11031 from zsxwing/awaitTermination.
Diffstat (limited to 'python/pyspark/streaming/tests.py')
| -rw-r--r-- | python/pyspark/streaming/tests.py | 4 |
1 files changed, 2 insertions, 2 deletions
diff --git a/python/pyspark/streaming/tests.py b/python/pyspark/streaming/tests.py index 24b812615c..b33e8252a7 100644 --- a/python/pyspark/streaming/tests.py +++ b/python/pyspark/streaming/tests.py @@ -1013,12 +1013,12 @@ class KafkaStreamTests(PySparkStreamingTestCase): self._kafkaTestUtils.setup() def tearDown(self): + super(KafkaStreamTests, self).tearDown() + if self._kafkaTestUtils is not None: self._kafkaTestUtils.teardown() self._kafkaTestUtils = None - super(KafkaStreamTests, self).tearDown() - def _randomTopic(self): return "topic-%d" % random.randint(0, 10000) |