You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/01/18 20:16:46 UTC
[GitHub] [spark] jerrypeng opened a new pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
jerrypeng opened a new pull request #35238:
URL: https://github.com/apache/spark/pull/35238
### What changes were proposed in this pull request?
To add support to using the AvailableNow trigger for the KafkaSource
### Why are the changes needed?
To allow users to use the KafkaSource the AvailableNow trigger. The AvailableNow trigger allows the all the data currently available in the source to be processed in multiple micro batches.
### Does this PR introduce _any_ user-facing change?
Yes, allows users to using the KafkaSource to use the AvailableNow trigger
### How was this patch tested?
Added a test for this.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r789075061
##########
File path: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
##########
@@ -195,6 +195,45 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
true
}
+ test("Trigger.AvailableNow") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 5)
+
+ testUtils.sendMessages(topic, (0 until 15).map { case x =>
+ s"foo-$x"
+ }.toArray, Some(0))
+
+ val reader = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("kafka.metadata.max.age.ms", "1")
+ .option("maxOffsetsPerTrigger", 5)
+ .option("subscribe", topic)
+ .option("startingOffsets", "earliest")
+ .load()
+
+ var index: Int = 0
+ def startTriggerAvailableNowQuery(): StreamingQuery = {
+ reader.writeStream
+ .foreachBatch((_: Dataset[Row], _: Long) => {
+ index+=1
+ })
+ .trigger(Trigger.AvailableNow)
+ .start()
+ }
+
+ val q1 = startTriggerAvailableNowQuery()
Review comment:
sure
##########
File path: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
##########
@@ -195,6 +195,45 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
true
}
+ test("Trigger.AvailableNow") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 5)
+
+ testUtils.sendMessages(topic, (0 until 15).map { case x =>
+ s"foo-$x"
+ }.toArray, Some(0))
+
+ val reader = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("kafka.metadata.max.age.ms", "1")
+ .option("maxOffsetsPerTrigger", 5)
+ .option("subscribe", topic)
+ .option("startingOffsets", "earliest")
+ .load()
+
+ var index: Int = 0
+ def startTriggerAvailableNowQuery(): StreamingQuery = {
+ reader.writeStream
+ .foreachBatch((_: Dataset[Row], _: Long) => {
+ index+=1
Review comment:
sure
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r788527444
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
##########
@@ -98,7 +100,10 @@ private[kafka010] class KafkaMicroBatchStream(
} else if (minOffsetPerTrigger.isDefined) {
ReadLimit.minRows(minOffsetPerTrigger.get, maxTriggerDelayMs)
} else {
- maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(super.getDefaultReadLimit)
+ // Calling ReadLimit.allAvailable() explicitly instead of super.getDefaultReadLimit because
+ // scala compiler bug https://github.com/scala/bug/issues/12523
+ // TODO revert to calling super.getDefaultReadLimit when scala complier bug is fixed
Review comment:
It is better to create a JIRA ticket and add into the comment here for tracking it.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng closed pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng closed pull request #35238:
URL: https://github.com/apache/spark/pull/35238
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r788529918
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSource.scala
##########
@@ -130,7 +134,10 @@ private[kafka010] class KafkaSource(
} else if (minOffsetPerTrigger.isDefined) {
ReadLimit.minRows(minOffsetPerTrigger.get, maxTriggerDelayMs)
} else {
- maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(super.getDefaultReadLimit)
+ // Calling ReadLimit.allAvailable() explicitly instead of super.getDefaultReadLimit because
+ // scala compiler bug https://github.com/scala/bug/issues/12523
+ // TODO revert to calling super.getDefaultReadLimit when scala complier bug is fixed
Review comment:
ditto.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35238: [SPARK-36649] - Support Trigger.AvailableNow on Kafka data source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r786689190
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSource.scala
##########
@@ -130,7 +134,7 @@ private[kafka010] class KafkaSource(
} else if (minOffsetPerTrigger.isDefined) {
ReadLimit.minRows(minOffsetPerTrigger.get, maxTriggerDelayMs)
} else {
- maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(super.getDefaultReadLimit)
+ maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(ReadLimit.allAvailable() )
Review comment:
nit: no space before )
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r788516739
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSource.scala
##########
@@ -77,7 +77,9 @@ private[kafka010] class KafkaSource(
metadataPath: String,
startingOffsets: KafkaOffsetRangeLimit,
failOnDataLoss: Boolean)
- extends SupportsAdmissionControl with Source with Logging {
+ extends SupportsTriggerAvailableNow
+ with Source
+ with Logging {
Review comment:
super nit: maybe still keep all the extends in one-line?
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
##########
@@ -98,7 +100,7 @@ private[kafka010] class KafkaMicroBatchStream(
} else if (minOffsetPerTrigger.isDefined) {
ReadLimit.minRows(minOffsetPerTrigger.get, maxTriggerDelayMs)
} else {
- maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(super.getDefaultReadLimit)
+ maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(ReadLimit.allAvailable())
Review comment:
how about fill a Spark Jira with the detailed description. Then we just need a single line comment here like:
```
TODO(SPARK-xxxx): directly call super.getDefaultReadLimit when ...
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] dongjoon-hyun commented on pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
dongjoon-hyun commented on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1018312616
+1, late LGTM. Thank you all!
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787240305
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
##########
@@ -98,7 +100,7 @@ private[kafka010] class KafkaMicroBatchStream(
} else if (minOffsetPerTrigger.isDefined) {
ReadLimit.minRows(minOffsetPerTrigger.get, maxTriggerDelayMs)
} else {
- maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(super.getDefaultReadLimit)
+ maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(ReadLimit.allAvailable())
Review comment:
https://github.com/apache/spark/pull/35238#discussion_r786502742
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r789082758
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
##########
@@ -98,7 +100,7 @@ private[kafka010] class KafkaMicroBatchStream(
} else if (minOffsetPerTrigger.isDefined) {
ReadLimit.minRows(minOffsetPerTrigger.get, maxTriggerDelayMs)
} else {
- maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(super.getDefaultReadLimit)
+ maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(ReadLimit.allAvailable())
Review comment:
sure
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787402655
##########
File path: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
##########
@@ -195,6 +195,45 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
true
}
+ test("Trigger.AvailableNow") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 5)
+
+ testUtils.sendMessages(topic, (0 until 15).map{ case x =>
+ s"foo-$x"
+ }.toArray, Some(0))
+
+ val reader = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("kafka.metadata.max.age.ms", "1")
+ .option("maxOffsetsPerTrigger", 5)
+ .option("subscribe", topic)
+ .option("startingOffsets", "earliest")
+ .load()
+
+ var index: Int = 0
+ def startTriggerAvailableNowQuery(): StreamingQuery = {
+ reader.writeStream
+ .foreachBatch{ case (_: Dataset[Row], _: Long) =>
Review comment:
My apologize for back and forth. I roughly remember this worked, but looks like it doesn't work. Please revert the latest change.
```
[error] /home/runner/work/spark/spark/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala:219:22: missing parameter type for expanded function
[error] The argument types of an anonymous function must be fully known. (SLS 8.5)
[error] Expected type was: ?
[error] .foreachBatch{ case (_: Dataset[Row], _: Long) =>
[error] ^
[error] one error found
```
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787211335
##########
File path: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
##########
@@ -195,6 +195,45 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
true
}
+ test("Trigger.AvailableNow") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 5)
+
+ testUtils.sendMessages(topic, (0 until 15).map(x => {
Review comment:
nit: we tend to compose ( and { if possible. For this case, you can use `{ case x =>` to get rid of (.
##########
File path: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
##########
@@ -195,6 +195,45 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
true
}
+ test("Trigger.AvailableNow") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 5)
+
+ testUtils.sendMessages(topic, (0 until 15).map(x => {
+ "foo-" + x
+ }).toArray, Some(0))
+
+ val reader = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("kafka.metadata.max.age.ms", "1")
+ .option("maxOffsetsPerTrigger", 5)
+ .option("subscribe", topic)
+ .option("startingOffsets", "earliest")
+ .load()
+
+ var index: Int = 0
+ def startTriggerAvailableNowQuery(): StreamingQuery = {
+ reader.writeStream
+ .foreachBatch((df: Dataset[Row], i: Long) => {
Review comment:
nit: same reason, please try `{ case (df: Dataset[Row], i: Long) =>` instead and remove (.
##########
File path: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
##########
@@ -195,6 +195,45 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
true
}
+ test("Trigger.AvailableNow") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 5)
+
+ testUtils.sendMessages(topic, (0 until 15).map(x => {
+ "foo-" + x
Review comment:
nit: We prefer to use string interpolation: s"foo-$x"
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r789074526
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSource.scala
##########
@@ -77,7 +77,9 @@ private[kafka010] class KafkaSource(
metadataPath: String,
startingOffsets: KafkaOffsetRangeLimit,
failOnDataLoss: Boolean)
- extends SupportsAdmissionControl with Source with Logging {
+ extends SupportsTriggerAvailableNow
+ with Source
+ with Logging {
Review comment:
sure
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #35238: [SPARK-36649] - Support Trigger.AvailableNow on Kafka data source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1015344848
cc. @xuanyuanking
Please take a look into this PR. Once approved, you can leverage this PR to test setup with merge script for Spark repo.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787117042
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSource.scala
##########
@@ -130,7 +134,7 @@ private[kafka010] class KafkaSource(
} else if (minOffsetPerTrigger.isDefined) {
ReadLimit.minRows(minOffsetPerTrigger.get, maxTriggerDelayMs)
} else {
- maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(super.getDefaultReadLimit)
+ maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(ReadLimit.allAvailable() )
Review comment:
fixed
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r788531139
##########
File path: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
##########
@@ -195,6 +195,45 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
true
}
+ test("Trigger.AvailableNow") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 5)
+
+ testUtils.sendMessages(topic, (0 until 15).map { case x =>
+ s"foo-$x"
+ }.toArray, Some(0))
+
+ val reader = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("kafka.metadata.max.age.ms", "1")
+ .option("maxOffsetsPerTrigger", 5)
+ .option("subscribe", topic)
+ .option("startingOffsets", "earliest")
+ .load()
+
+ var index: Int = 0
+ def startTriggerAvailableNowQuery(): StreamingQuery = {
+ reader.writeStream
+ .foreachBatch((_: Dataset[Row], _: Long) => {
+ index+=1
Review comment:
style: `index += 1`
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r788532598
##########
File path: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
##########
@@ -195,6 +195,45 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
true
}
+ test("Trigger.AvailableNow") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 5)
+
+ testUtils.sendMessages(topic, (0 until 15).map { case x =>
+ s"foo-$x"
+ }.toArray, Some(0))
+
+ val reader = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("kafka.metadata.max.age.ms", "1")
+ .option("maxOffsetsPerTrigger", 5)
+ .option("subscribe", topic)
+ .option("startingOffsets", "earliest")
+ .load()
+
+ var index: Int = 0
+ def startTriggerAvailableNowQuery(): StreamingQuery = {
+ reader.writeStream
+ .foreachBatch((_: Dataset[Row], _: Long) => {
+ index+=1
+ })
+ .trigger(Trigger.AvailableNow)
+ .start()
+ }
+
+ val q1 = startTriggerAvailableNowQuery()
Review comment:
nit: just `query`? there is only one query.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r788534527
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
##########
@@ -57,7 +57,7 @@ private[kafka010] class KafkaMicroBatchStream(
metadataPath: String,
startingOffsets: KafkaOffsetRangeLimit,
failOnDataLoss: Boolean)
- extends SupportsAdmissionControl with ReportsSourceMetrics with MicroBatchStream with Logging {
+ extends SupportsTriggerAvailableNow with ReportsSourceMetrics with MicroBatchStream with Logging {
Review comment:
`SupportsTriggerAvailableNow` already extends `SupportsAdmissionControl`:
https://github.com/apache/spark/blob/master/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/streaming/SupportsTriggerAvailableNow.java
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787281722
##########
File path: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
##########
@@ -195,6 +195,45 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
true
}
+ test("Trigger.AvailableNow") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 5)
+
+ testUtils.sendMessages(topic, (0 until 15).map(x => {
+ "foo-" + x
+ }).toArray, Some(0))
+
+ val reader = spark
+ .readStream
+ .format("kafka")
+ .option("kafka.bootstrap.servers", testUtils.brokerAddress)
+ .option("kafka.metadata.max.age.ms", "1")
+ .option("maxOffsetsPerTrigger", 5)
+ .option("subscribe", topic)
+ .option("startingOffsets", "earliest")
+ .load()
+
+ var index: Int = 0
+ def startTriggerAvailableNowQuery(): StreamingQuery = {
+ reader.writeStream
+ .foreachBatch((df: Dataset[Row], i: Long) => {
Review comment:
sure I can change this I just this copied format from existing code
https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/streaming/FileStreamSourceSuite.scala#L1354
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787280362
##########
File path: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
##########
@@ -195,6 +195,45 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
true
}
+ test("Trigger.AvailableNow") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 5)
+
+ testUtils.sendMessages(topic, (0 until 15).map(x => {
Review comment:
sure I can change this I was just this copied format from existing code
https://github.com/apache/spark/blob/master/external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala#L253
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787301722
##########
File path: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
##########
@@ -199,9 +199,9 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
val topic = newTopic()
testUtils.createTopic(topic, partitions = 5)
- testUtils.sendMessages(topic, (0 until 15).map(x => {
- "foo-" + x
- }).toArray, Some(0))
+ testUtils.sendMessages(topic, (0 until 15).map{ case x =>
Review comment:
sorry for the nits: space between `p` and `{` / same happens in foreachBatch
Not sure style checker will check this, but it is good practice to run`dev/scalastyle` before pushing the change. Not all styles are checked with style checker unfortunately.
Please go through https://github.com/databricks/scala-style-guide when you have some time.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on a change in pull request #35238: [SPARK-36649] - Support Trigger.AvailableNow on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r786502742
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSource.scala
##########
@@ -130,7 +134,7 @@ private[kafka010] class KafkaSource(
} else if (minOffsetPerTrigger.isDefined) {
ReadLimit.minRows(minOffsetPerTrigger.get, maxTriggerDelayMs)
} else {
- maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(super.getDefaultReadLimit)
+ maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(ReadLimit.allAvailable() )
Review comment:
Had to make this change because of scala compiler bug that I encountered:
https://github.com/scala/bug/issues/12523
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787249747
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
##########
@@ -57,7 +57,7 @@ private[kafka010] class KafkaMicroBatchStream(
metadataPath: String,
startingOffsets: KafkaOffsetRangeLimit,
failOnDataLoss: Boolean)
- extends SupportsAdmissionControl with ReportsSourceMetrics with MicroBatchStream with Logging {
+ extends SupportsTriggerAvailableNow with ReportsSourceMetrics with MicroBatchStream with Logging {
Review comment:
I don't think it is necessary. It is redundant
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking closed pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
xuanyuanking closed pull request #35238:
URL: https://github.com/apache/spark/pull/35238
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng removed a comment on pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng removed a comment on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1015799391
> Welcome to Spark community and thanks for the contribution @jerrypeng !
>
> Looks like the test you added failed in Github Action. https://github.com/jerrypeng/spark/runs/4850411860?check_suite_focus=true
>
> Could you please look into it?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1015799391
> Welcome to Spark community and thanks for the contribution @jerrypeng !
>
> Looks like the test you added failed in Github Action. https://github.com/jerrypeng/spark/runs/4850411860?check_suite_focus=true
>
> Could you please look into it?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] alex-balikov commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
alex-balikov commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787222535
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
##########
@@ -57,7 +57,7 @@ private[kafka010] class KafkaMicroBatchStream(
metadataPath: String,
startingOffsets: KafkaOffsetRangeLimit,
failOnDataLoss: Boolean)
- extends SupportsAdmissionControl with ReportsSourceMetrics with MicroBatchStream with Logging {
+ extends SupportsTriggerAvailableNow with ReportsSourceMetrics with MicroBatchStream with Logging {
Review comment:
what about SupportsAdmissionControl - I see other APIs specifying both - https://livegrep.dev.databricks.com/view/databricks/runtime/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/FileStreamSource.scala#L57
even if one extends the other, is it better to list them explicitly?
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
##########
@@ -98,7 +100,7 @@ private[kafka010] class KafkaMicroBatchStream(
} else if (minOffsetPerTrigger.isDefined) {
ReadLimit.minRows(minOffsetPerTrigger.get, maxTriggerDelayMs)
} else {
- maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(super.getDefaultReadLimit)
+ maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(ReadLimit.allAvailable())
Review comment:
why this change? Isn't it better to delegate to the parent if you have no opinion - e.g. the parent could use some config default or whatever.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787240136
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
##########
@@ -98,7 +100,7 @@ private[kafka010] class KafkaMicroBatchStream(
} else if (minOffsetPerTrigger.isDefined) {
ReadLimit.minRows(minOffsetPerTrigger.get, maxTriggerDelayMs)
} else {
- maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(super.getDefaultReadLimit)
+ maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(ReadLimit.allAvailable())
Review comment:
Previous comment seems to be out of date. It's to avoid Scala compiler bug.
While we are here, probably good to comment for the reason, and tag `// TODO: `
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r788534527
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
##########
@@ -57,7 +57,7 @@ private[kafka010] class KafkaMicroBatchStream(
metadataPath: String,
startingOffsets: KafkaOffsetRangeLimit,
failOnDataLoss: Boolean)
- extends SupportsAdmissionControl with ReportsSourceMetrics with MicroBatchStream with Logging {
+ extends SupportsTriggerAvailableNow with ReportsSourceMetrics with MicroBatchStream with Logging {
Review comment:
`SupportsTriggerAvailableNow` extends `SupportsAdmissionControl`:
https://github.com/apache/spark/blob/master/sql/catalyst/src/main/java/org/apache/spark/sql/connector/read/streaming/SupportsTriggerAvailableNow.java
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1017849844
Will merge this right after the test passes. Thanks for the contribution and welcome to the community! @jerrypeng
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1018276741
@xuanyuanking Friendly reminder. Please go with merging at your convenience. Thanks!
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1017846245
@viirya and @xuanyuanking thank you for the review. I have addressed your comments.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #35238: [SPARK-36649] - Support Trigger.AvailableNow on Kafka data source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1015343308
Welcome to Spark community and thanks for the contribution @jerrypeng !
Looks like the test you added failed in Github Action.
https://github.com/jerrypeng/spark/runs/4850411860?check_suite_focus=true
Could you please look into it?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787281819
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
##########
@@ -98,7 +100,7 @@ private[kafka010] class KafkaMicroBatchStream(
} else if (minOffsetPerTrigger.isDefined) {
ReadLimit.minRows(minOffsetPerTrigger.get, maxTriggerDelayMs)
} else {
- maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(super.getDefaultReadLimit)
+ maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(ReadLimit.allAvailable())
Review comment:
I will add a comment
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787281480
##########
File path: external/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchSourceSuite.scala
##########
@@ -195,6 +195,45 @@ abstract class KafkaMicroBatchSourceSuiteBase extends KafkaSourceSuiteBase {
true
}
+ test("Trigger.AvailableNow") {
+ val topic = newTopic()
+ testUtils.createTopic(topic, partitions = 5)
+
+ testUtils.sendMessages(topic, (0 until 15).map(x => {
+ "foo-" + x
Review comment:
sure will change
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng edited a comment on pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng edited a comment on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1015799916
@HeartSaVioR thank you! Glad to join the Spark community!
I have added the fixes for the failed test.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1015997129
@HeartSaVioR @alex-balikov thank you for the review. I have address your comments. Please take another look.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1015796061
Can one of the admins verify this patch?
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] jerrypeng commented on pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1015799916
@HeartSaVioR added the fixes
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #35238:
URL: https://github.com/apache/spark/pull/35238#discussion_r787240136
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala
##########
@@ -98,7 +100,7 @@ private[kafka010] class KafkaMicroBatchStream(
} else if (minOffsetPerTrigger.isDefined) {
ReadLimit.minRows(minOffsetPerTrigger.get, maxTriggerDelayMs)
} else {
- maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(super.getDefaultReadLimit)
+ maxOffsetsPerTrigger.map(ReadLimit.maxRows).getOrElse(ReadLimit.allAvailable())
Review comment:
Previous comment seems to be out of date. It's to avoid Scala compiler bug.
While we are here, probably good to leave code comment for the reason, and tag `// TODO: ` to fix it eventually.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on pull request #35238: [SPARK-36649][SQL] Support `Trigger.AvailableNow` on Kafka data source
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on pull request #35238:
URL: https://github.com/apache/spark/pull/35238#issuecomment-1018316258
Merged to master.
--
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.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org