HeartSaVioR commented on a change in pull request #26292: [SPARK-29635][SS]
Extract base test suites between Kafka micro-batch sink and Kafka continuous
sink
URL: https://github.com/apache/spark/pull/26292#discussion_r339956557
##########
File path:
external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaSinkSuite.scala
##########
@@ -97,159 +108,123 @@ class KafkaSinkStreamingSuite extends
KafkaSinkSuiteBase with StreamTest {
.map(_._2)
try {
- input.addData("1", "2", "3", "4", "5")
- failAfter(streamingTimeout) {
- writer.processAllAvailable()
- }
- checkDatasetUnorderly(reader, 1, 2, 3, 4, 5)
- input.addData("6", "7", "8", "9", "10")
- failAfter(streamingTimeout) {
- writer.processAllAvailable()
- }
- checkDatasetUnorderly(reader, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
+ input.addData(1, 2, 3, 4, 5)
+ verifyResult(writer)(checkDatasetUnorderly(reader, 1, 2, 3, 4, 5))
+ input.addData(6, 7, 8, 9, 10)
+ verifyResult(writer)(checkDatasetUnorderly(reader, 1, 2, 3, 4, 5, 6, 7,
8, 9, 10))
} finally {
writer.stop()
}
}
- test("streaming - write aggregation w/o topic field, with topic option") {
- val input = MemoryStream[String]
+ test("streaming - write w/o topic field, with topic option") {
+ val input = createMemoryStream()
val topic = newTopic()
testUtils.createTopic(topic)
val writer = createKafkaWriter(
- input.toDF().groupBy("value").count(),
+ input.toDF(),
withTopic = Some(topic),
- withOutputMode = Some(OutputMode.Update()))(
- withSelectExpr = "CAST(value as STRING) key", "CAST(count as STRING)
value")
+ withOutputMode = Some(OutputMode.Append()))(
+ withSelectExpr = "CAST(value as STRING) value")
val reader = createKafkaReader(topic)
.selectExpr("CAST(key as STRING) key", "CAST(value as STRING) value")
.selectExpr("CAST(key as INT) key", "CAST(value as INT) value")
- .as[(Int, Int)]
+ .as[(Option[Int], Int)]
+ .map(_._2)
try {
- input.addData("1", "2", "2", "3", "3", "3")
- failAfter(streamingTimeout) {
- writer.processAllAvailable()
- }
- checkDatasetUnorderly(reader, (1, 1), (2, 2), (3, 3))
- input.addData("1", "2", "3")
- failAfter(streamingTimeout) {
- writer.processAllAvailable()
- }
- checkDatasetUnorderly(reader, (1, 1), (2, 2), (3, 3), (1, 2), (2, 3),
(3, 4))
+ input.addData(1, 2, 3, 4, 5)
+ verifyResult(writer)(checkDatasetUnorderly(reader, 1, 2, 3, 4, 5))
+ input.addData(6, 7, 8, 9, 10)
+ verifyResult(writer)(checkDatasetUnorderly(reader, 1, 2, 3, 4, 5, 6, 7,
8, 9, 10))
} finally {
writer.stop()
}
}
- test("streaming - aggregation with topic field and topic option") {
Review comment:
Same here.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
