kavpreetgrewal commented on code in PR #59219:
URL: https://github.com/apache/spark/pull/59219#discussion_r4170784500
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala:
##########
@@ -374,13 +374,16 @@ private[kafka010] class KafkaSourceProvider extends
DataSourceRegister
STARTING_OFFSETS_OPTION_KEY, EarliestOffsetRangeLimit) match {
case EarliestOffsetRangeLimit => // good to go
case LatestOffsetRangeLimit =>
- throw new IllegalArgumentException("starting offset can't be latest " +
- "for batch queries on Kafka")
+ throw KafkaExceptions.invalidBatchOffset(
+ offsetName = STARTING_OFFSETS_OPTION_KEY,
+ offsetValue = "latest")
case SpecificOffsetRangeLimit(partitionOffsets, topicOffsets) =>
(partitionOffsets.map { case (tp, off) => tp.toString -> off } ++
topicOffsets).foreach {
case (name, off) if off == KafkaOffsetRangeLimit.LATEST =>
- throw new IllegalArgumentException(s"startingOffsets for $name
can't " +
- "be latest for batch queries on Kafka")
+ throw KafkaExceptions.invalidBatchOffsetForTopicOrPartition(
+ offsetName = STARTING_OFFSETS_OPTION_KEY,
+ offsetValue = "latest",
+ topicOrPartition = name)
Review Comment:
Technically this is the topic _and_ partition so this should be named
`topicPartition`
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
