Skip to content

[SPARK-25644][SS][FOLLOWUP][BUILD] Fix Scala 2.12 build error due to foreachBatch #22649

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
dongjoon-hyun wants to merge 1 commit into from
Closed

[SPARK-25644][SS][FOLLOWUP][BUILD] Fix Scala 2.12 build error due to foreachBatch #22649

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
4 changes: 2 additions & 2 deletions ...ka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ import org.apache.kafka.common.TopicPartition
import org.scalatest.concurrent.PatienceConfiguration.Timeout
import org.scalatest.time.SpanSugar._

import org.apache.spark.sql.{ForeachWriter, SparkSession}
import org.apache.spark.sql.{Dataset, ForeachWriter, SparkSession}
import org.apache.spark.sql.execution.datasources.v2.StreamingDataSourceV2Relation
import org.apache.spark.sql.execution.exchange.ReusedExchangeExec
import org.apache.spark.sql.execution.streaming._
Expand Down Expand Up @@ -900,7 +900,7 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
}
testUtils.waitUntilOffsetAppears(topicPartition, 5)

val q = ds.writeStream.foreachBatch { (ds, epochId) =>
val q = ds.writeStream.foreachBatch { (ds: Dataset[String], epochId: Long) =>
if (epochId == 0) {
// Send more message before the tasks of the current batch start reading the current batch
// data, so that the executors will prefetch messages in the next batch and drop them. In
Expand Down
5 changes: 3 additions & 2 deletions ...c/test/scala/org/apache/spark/sql/execution/streaming/sources/ForeachBatchSinkSuite.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,11 +99,12 @@ class ForeachBatchSinkSuite extends StreamTest {
}
assert(ex1.getMessage.contains("foreachBatch function cannot be null"))
val ex2 = intercept[AnalysisException] {
ds.writeStream.foreachBatch((_, _) => {}).trigger(Trigger.Continuous("1 second")).start()
ds.writeStream.foreachBatch((_: Dataset[Int], _: Long) => {})
.trigger(Trigger.Continuous("1 second")).start()
}
assert(ex2.getMessage.contains("'foreachBatch' is not supported with continuous trigger"))
val ex3 = intercept[AnalysisException] {
ds.writeStream.foreachBatch((_, _) => {}).partitionBy("value").start()
ds.writeStream.foreachBatch((_: Dataset[Int], _: Long) => {}).partitionBy("value").start()
}
assert(ex3.getMessage.contains("'foreachBatch' does not support partitioning"))
}
Expand Down