kavpreetgrewal commented on code in PR #58043:
URL: https://github.com/apache/spark/pull/58043#discussion_r3826715849
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala:
##########
@@ -155,8 +155,10 @@ private[kafka010] class KafkaOffsetReaderAdmin(
}
override def fetchSpecificOffsets(
- partitionOffsets: Map[TopicPartition, Long],
+ offsets: SpecificOffsetRangeLimit,
reportDataLoss: (String, () => Throwable) => Unit): KafkaSourceOffset = {
+ val partitionOffsets = offsets.resolve(withRetries { resolvePartitions() })
Review Comment:
We can resolve the topic-level offsets inside `fetchSpecificOffsets0` using
the callback’s partitions. If we don't, then the metadata can change between
pre-resolution and fetch, which would make the valid topic-level offsets fail
the partition assertion. Same applies in the consumer reader.
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/JsonUtils.scala:
##########
@@ -76,6 +79,43 @@ private object JsonUtils {
}
}
+ /**
+ * Read the offsets of the `startingOffsets` / `endingOffsets` json string.
On top of the
+ * per-TopicPartition form read by [[partitionOffsets]], a topic may bind to
"earliest" or
+ * "latest" as a whole, e.g. {"topicA":"earliest","topicB":{"0":23,"1":-1}}.
Such topic-level
+ * values are returned unexpanded in
`SpecificOffsetRangeLimit.topicOffsets`, since the
+ * partitions of the topic are only known once the offsets get resolved
against Kafka.
+ */
+ def specificOffsets(str: String): SpecificOffsetRangeLimit = {
+ def fail(): Nothing = throw new IllegalArgumentException(
+ s"""Expected e.g. {"topicA":{"0":23,"1":-1},"topicB":{"0":-2}} or
+ |{"topicA":"earliest","topicB":"latest"}, got $str""".stripMargin)
+
+ val partitionOffsets = new HashMap[TopicPartition, Long]
+ val topicOffsets = new HashMap[String, Long]
+ try {
+ parse(str) match {
+ case JObject(topics) =>
+ topics.foreach {
+ case (topic, JString(value)) =>
+ topicOffsets += topic -> (value.toLowerCase(Locale.ROOT) match {
+ case "earliest" => KafkaOffsetRangeLimit.EARLIEST
+ case "latest" => KafkaOffsetRangeLimit.LATEST
+ case _ => fail()
+ })
+ case (topic, partOffsets) =>
+ partOffsets.extract[Map[Int, Long]].foreach { case (part,
offset) =>
+ partitionOffsets += new TopicPartition(topic, part) -> offset
+ }
+ }
+ case _ => fail()
+ }
+ } catch {
+ case NonFatal(_) => fail()
+ }
+ SpecificOffsetRangeLimit(partitionOffsets.toMap, topicOffsets.toMap)
Review Comment:
Can we have a check for when a topic is in both topicOffsets and
partitionOffsets? We should throw an error in that case
##########
docs/streaming/structured-streaming-kafka-integration.md:
##########
@@ -442,13 +451,15 @@ The following configurations are optional:
<tr>
<td>endingOffsets</td>
<td>latest or json string
- {"topicA":{"0":23,"1":-1},"topicB":{"0":-1}}
+ {"topicA":{"0":23,"1":-1},"topicB":{"0":-1}} or
{"topicA":"latest","topicB":"latest"}
Review Comment:
Can you add the example, where it has both the string and the per-partition
values? It makes it more explicit that both work, not just either or
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala:
##########
@@ -375,13 +375,19 @@ private[kafka010] class KafkaSourceProvider extends
DataSourceRegister
case LatestOffsetRangeLimit =>
throw new IllegalArgumentException("starting offset can't be latest " +
"for batch queries on Kafka")
- case SpecificOffsetRangeLimit(partitionOffsets) =>
+ case SpecificOffsetRangeLimit(partitionOffsets, topicOffsets) =>
partitionOffsets.foreach {
case (tp, off) if off == KafkaOffsetRangeLimit.LATEST =>
throw new IllegalArgumentException(s"startingOffsets for $tp can't
" +
"be latest for batch queries on Kafka")
case _ => // ignore
}
+ topicOffsets.foreach {
Review Comment:
nit: can we combine both of these foreach? same for the block below
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReaderAdmin.scala:
##########
@@ -155,8 +155,10 @@ private[kafka010] class KafkaOffsetReaderAdmin(
}
override def fetchSpecificOffsets(
- partitionOffsets: Map[TopicPartition, Long],
+ offsets: SpecificOffsetRangeLimit,
reportDataLoss: (String, () => Throwable) => Unit): KafkaSourceOffset = {
+ val partitionOffsets = offsets.resolve(withRetries { resolvePartitions() })
Review Comment:
Can we move `withRetries { resolvePartitions() }` into a helper
`fetchTopicPartitions`. The consumer reader already has this so the method can
be defined in the super class as well
##########
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/JsonUtils.scala:
##########
Review Comment:
Is this still used anywhere? Or can it be combined with specific 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: [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]
