[GitHub] [spark] jerrypeng commented on a diff in pull request #38911: [SPARK-41387][SS] Assert current end offset from Kafka data source for Trigger.AvailableNow
jerrypeng commented on code in PR #38911:
URL: https://github.com/apache/spark/pull/38911#discussion_r1042793780
##
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala:
##
@@ -316,6 +320,50 @@ private[kafka010] class KafkaMicroBatchStream(
}
}
+ private def assertEndOffsetForTriggerAvailableNow(
+ endPartitionOffsets: Map[TopicPartition, Long]): Unit = {
+val tpsForPrefetched = allDataForTriggerAvailableNow.keySet
+val tpsForEndOffset = endPartitionOffsets.keySet
+
+if (tpsForPrefetched != tpsForEndOffset) {
+ throw KafkaExceptions.topicPartitionsInEndOffsetAreNotSameWithPrefetched(
+tpsForPrefetched, tpsForEndOffset)
+}
+
+val endOffsetHasGreaterThanPrefetched = {
+ allDataForTriggerAvailableNow.keySet.exists { tp =>
+val offsetFromPrefetched = allDataForTriggerAvailableNow(tp)
+val offsetFromEndOffset = endPartitionOffsets(tp)
+offsetFromEndOffset > offsetFromPrefetched
+ }
+}
+if (endOffsetHasGreaterThanPrefetched) {
+ throw
KafkaExceptions.endOffsetHasGreaterOffsetForTopicPartitionThanPrefetched(
+allDataForTriggerAvailableNow, endPartitionOffsets)
+}
+
+val latestOffsets =
kafkaOffsetReader.fetchLatestOffsets(Some(endPartitionOffsets))
Review Comment:
These things can happen regardless of whether availableNow trigger is used
right? What happens if these things happen when using processing time trigger?
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] jerrypeng commented on a diff in pull request #38911: [SPARK-41387][SS] Assert current end offset from Kafka data source for Trigger.AvailableNow
jerrypeng commented on code in PR #38911:
URL: https://github.com/apache/spark/pull/38911#discussion_r1042792623
##
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala:
##
@@ -194,6 +194,10 @@ private[kafka010] class KafkaMicroBatchStream(
val startPartitionOffsets =
start.asInstanceOf[KafkaSourceOffset].partitionToOffsets
val endPartitionOffsets =
end.asInstanceOf[KafkaSourceOffset].partitionToOffsets
+if (allDataForTriggerAvailableNow != null) {
Review Comment:
Can we do a sanity check i.e. the end topic offsets are larger than start
topic offsets?
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] jerrypeng commented on a diff in pull request #38911: [SPARK-41387][SS] Assert current end offset from Kafka data source for Trigger.AvailableNow
jerrypeng commented on code in PR #38911:
URL: https://github.com/apache/spark/pull/38911#discussion_r1041856979
##
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaMicroBatchStream.scala:
##
@@ -316,6 +320,50 @@ private[kafka010] class KafkaMicroBatchStream(
}
}
+ private def assertEndOffsetForTriggerAvailableNow(
+ endPartitionOffsets: Map[TopicPartition, Long]): Unit = {
+val tpsForPrefetched = allDataForTriggerAvailableNow.keySet
+val tpsForEndOffset = endPartitionOffsets.keySet
+
+if (tpsForPrefetched != tpsForEndOffset) {
+ throw KafkaExceptions.topicPartitionsInEndOffsetAreNotSameWithPrefetched(
+tpsForPrefetched, tpsForEndOffset)
+}
+
+val endOffsetHasGreaterThanPrefetched = {
+ allDataForTriggerAvailableNow.keySet.exists { tp =>
+val offsetFromPrefetched = allDataForTriggerAvailableNow(tp)
+val offsetFromEndOffset = endPartitionOffsets(tp)
+offsetFromEndOffset > offsetFromPrefetched
+ }
+}
+if (endOffsetHasGreaterThanPrefetched) {
+ throw
KafkaExceptions.endOffsetHasGreaterOffsetForTopicPartitionThanPrefetched(
+allDataForTriggerAvailableNow, endPartitionOffsets)
+}
+
+val latestOffsets =
kafkaOffsetReader.fetchLatestOffsets(Some(endPartitionOffsets))
Review Comment:
Why do we need to fetch the latest offset from Kafka again? This will add
additional latency. I know its trigger available now but what does this buy us?
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] jerrypeng commented on a diff in pull request #38911: [SPARK-41387][SS] Assert current end offset from Kafka data source for Trigger.AvailableNow
jerrypeng commented on code in PR #38911:
URL: https://github.com/apache/spark/pull/38911#discussion_r1041848809
##
connector/kafka-0-10-sql/src/test/resources/error/kafka-error-classes.json:
##
@@ -0,0 +1,26 @@
+{
+ "TOPIC_PARTITIONS_IN_END_OFFSET_ARE_NOT_SAME_WITH_PREFETCHED" : {
+"message" : [
+ "Kafka data source in Trigger.AvailableNow should provide the same topic
partitions in pre-fetched offset to end offset for each microbatch. ",
+ "topic-partitions for pre-fetched offset: ,
topic-partitions for end offset: ."
+]
+ },
+ "END_OFFSET_HAS_GREATER_OFFSET_FOR_TOPIC_PARTITION_THAN_PREFETCHED" : {
+"message" : [
+ "For Kafka data source with Trigger.AvailableNow, end offset should have
lower or equal offset per each topic partition than pre-fetched offset.",
+ "pre-fetched offset: , end offset: ."
+]
+ },
+ "LOST_TOPIC_PARTITIONS_IN_END_OFFSET_WITH_TRIGGER_AVAILABLENOW" : {
+"message" : [
+ "Some of partitions in Kafka topic(s) have been lost during running
query with Trigger.AvailableNow. Make sure topic partitions are not dropped
during the query run.",
Review Comment:
Same here. Transient error right?
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
[GitHub] [spark] jerrypeng commented on a diff in pull request #38911: [SPARK-41387][SS] Assert current end offset from Kafka data source for Trigger.AvailableNow
jerrypeng commented on code in PR #38911:
URL: https://github.com/apache/spark/pull/38911#discussion_r1041848355
##
connector/kafka-0-10-sql/src/test/resources/error/kafka-error-classes.json:
##
@@ -0,0 +1,26 @@
+{
+ "TOPIC_PARTITIONS_IN_END_OFFSET_ARE_NOT_SAME_WITH_PREFETCHED" : {
+"message" : [
+ "Kafka data source in Trigger.AvailableNow should provide the same topic
partitions in pre-fetched offset to end offset for each microbatch. ",
Review Comment:
This should be a transient issue right? As in when the job is triggered
again, the job should run fine right and number the job will read from
whichever partitions exist? Can you add that to this error message?
--
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-unsubscr...@spark.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
-
To unsubscribe, e-mail: reviews-unsubscr...@spark.apache.org
For additional commands, e-mail: reviews-h...@spark.apache.org
