This is an automated email from the ASF dual-hosted git repository.
dongjoon pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 18b86068ff4c [SPARK-47154][SS][TESTS] Fix `kafka-0-10-sql` to use
`ResetSystemProperties` if `KafkaTestUtils` is used
18b86068ff4c is described below
commit 18b86068ff4c72ba686d3d9275f9284d58cd3ef4
Author: Dongjoon Hyun <[email protected]>
AuthorDate: Sat Feb 24 11:15:05 2024 -0800
[SPARK-47154][SS][TESTS] Fix `kafka-0-10-sql` to use
`ResetSystemProperties` if `KafkaTestUtils` is used
### What changes were proposed in this pull request?
This PR aims to fix `kafka-0-10-sql` module to use `ResetSystemProperties`
if `KafkaTestUtils` is used. The following test suites are fixed.
- ConsumerStrategySuite
- KafkaDataConsumerSuite
- KafkaMissingOffsetsTest
- KafkaDontFailOnDataLossSuite
- KafkaSourceStressForDontFailOnDataLossSuite
- KafkaTest
- KafkaDelegationTokenSuite
- KafkaMicroBatchSourceSuite
- KafkaMicroBatchV1SourceWithAdminSuite
- KafkaMicroBatchV2SourceWithAdminSuite
- KafkaMicroBatchV1SourceSuite
- KafkaMicroBatchV2SourceSuite
- KafkaSourceStressSuite
- KafkaOffsetReaderSuite
- KafkaRelationSuite
- KafkaRelationSuiteWithAdminV1
- KafkaRelationSuiteWithAdminV2
- KafkaRelationSuiteV1
- KafkaRelationSuiteV2
- KafkaSinkSuite
- KafkaSinkMicroBatchStreamingSuite
- KafkaContinuousSinkSuite
- KafkaSinkBatchSuiteV1
- KafkaSinkBatchSuiteV2
### Why are the changes needed?
Apache Spark `master` branch has two `KafkaTestUtils` classes.
```
$ find . -name KafkaTestUtils.scala
./connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala
./connector/kafka-0-10/src/test/scala/org/apache/spark/streaming/kafka010/KafkaTestUtils.scala
```
`KafkaTestUtils` of `kafka-0-10-sql` uses `System.setProperty` and affects
8 files. We need to use `ResetSystemProperties` to isolate the test cases.
https://github.com/apache/spark/blob/ee312ecb40ea5b5303fc794a3d494b6f27cda923/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala#L290
```
$ git grep KafkaTestUtils connector/kafka-0-10-sql | awk -F: '{print $1}' |
sort | uniq
connector/kafka-0-10-sql/src/test/resources/log4j2.properties
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/ConsumerStrategySuite.scala
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDelegationTokenSuite.scala
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDontFailOnDataLossSuite.scala
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderSuite.scala
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaRelationSuite.scala
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSinkSuite.scala
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala
connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumerSuite.scala
```
### Does this PR introduce _any_ user-facing change?
No. This is a test-only PR.
### How was this patch tested?
Pass the CIs.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #45239 from dongjoon-hyun/SPARK-47154.
Authored-by: Dongjoon Hyun <[email protected]>
Signed-off-by: Dongjoon Hyun <[email protected]>
---
.../scala/org/apache/spark/sql/kafka010/ConsumerStrategySuite.scala | 3 ++-
.../org/apache/spark/sql/kafka010/KafkaDontFailOnDataLossSuite.scala | 3 ++-
.../src/test/scala/org/apache/spark/sql/kafka010/KafkaTest.scala | 3 ++-
.../apache/spark/sql/kafka010/consumer/KafkaDataConsumerSuite.scala | 2 ++
4 files changed, 8 insertions(+), 3 deletions(-)
diff --git
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/ConsumerStrategySuite.scala
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/ConsumerStrategySuite.scala
index 44baab7f2468..cbbbcf9317cd 100644
---
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/ConsumerStrategySuite.scala
+++
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/ConsumerStrategySuite.scala
@@ -27,8 +27,9 @@ import org.apache.kafka.common.TopicPartition
import org.mockito.Mockito.mock
import org.apache.spark.{SparkConf, SparkEnv, SparkFunSuite}
+import org.apache.spark.util.ResetSystemProperties
-class ConsumerStrategySuite extends SparkFunSuite {
+class ConsumerStrategySuite extends SparkFunSuite with ResetSystemProperties {
private var testUtils: KafkaTestUtils = _
private def doReturn(value: Any) = org.mockito.Mockito.doReturn(value,
Seq.empty: _*)
diff --git
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDontFailOnDataLossSuite.scala
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDontFailOnDataLossSuite.scala
index 2e0690f2288a..b46d2d2d0624 100644
---
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDontFailOnDataLossSuite.scala
+++
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDontFailOnDataLossSuite.scala
@@ -31,6 +31,7 @@ import org.apache.spark.sql.{DataFrame, Dataset,
ForeachWriter}
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.streaming.{StreamTest, Trigger}
import org.apache.spark.sql.test.{SharedSparkSession, TestSparkSession}
+import org.apache.spark.util.ResetSystemProperties
/**
* This is a basic test trait which will set up a Kafka cluster that keeps
only several records in
@@ -43,7 +44,7 @@ import org.apache.spark.sql.test.{SharedSparkSession,
TestSparkSession}
* does see missing offsets, you can check the earliest offset in `eventually`
and make sure it's
* not 0 rather than sleeping a hard-code duration.
*/
-trait KafkaMissingOffsetsTest extends SharedSparkSession {
+trait KafkaMissingOffsetsTest extends SharedSparkSession with
ResetSystemProperties {
protected var testUtils: KafkaTestUtils = _
diff --git
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTest.scala
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTest.scala
index 087d938f8ed8..194b3c528db8 100644
---
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTest.scala
+++
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTest.scala
@@ -21,9 +21,10 @@ import org.scalatest.BeforeAndAfterAll
import org.apache.spark.SparkFunSuite
import org.apache.spark.sql.kafka010.producer.InternalKafkaProducerPool
+import org.apache.spark.util.ResetSystemProperties
/** A trait to clean cached Kafka producers in `afterAll` */
-trait KafkaTest extends BeforeAndAfterAll {
+trait KafkaTest extends BeforeAndAfterAll with ResetSystemProperties {
self: SparkFunSuite =>
override def afterAll(): Unit = {
diff --git
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumerSuite.scala
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumerSuite.scala
index d4944334c2bf..b6748d0f261e 100644
---
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumerSuite.scala
+++
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/consumer/KafkaDataConsumerSuite.scala
@@ -35,10 +35,12 @@ import org.apache.spark.kafka010.KafkaDelegationTokenTest
import org.apache.spark.sql.kafka010.{KafkaTestUtils, RecordBuilder}
import org.apache.spark.sql.kafka010.consumer.KafkaDataConsumer.CacheKey
import org.apache.spark.sql.test.SharedSparkSession
+import org.apache.spark.util.ResetSystemProperties
class KafkaDataConsumerSuite
extends SharedSparkSession
with PrivateMethodTester
+ with ResetSystemProperties
with KafkaDelegationTokenTest {
protected var testUtils: KafkaTestUtils = _
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
