Kino1994 commented on code in PR #58043:
URL: https://github.com/apache/spark/pull/58043#discussion_r3832332601
##########
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:
Good catch, you were right. Resolving up front left a window in which the
partition metadata could change before the fetch, so valid topic-level offsets
could fail the assertion — exactly in the repartitioning scenario this feature
is meant to help with. Both readers now expand inside the callbacks, against
the partitions the fetch is actually run with, so the assertion and the
expansion always see the same set.
While doing this I also changed `fnAssertFetchedOffsets` in the consumer
reader to iterate `offsets.partitionOffsets` instead of the expanded map. Only
explicitly specified offsets have an expected value to compare against; the
expanded ones are always the earliest/latest sentinels, which the guard skipped
anyway.
##########
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:
Done. `fetchTopicPartitions()` is now declared in `KafkaOffsetReaderBase`
and implemented in both readers, with `withRetries { resolvePartitions() }` in
the admin one. The consumer implementation is unchanged apart from becoming an
override.
Note that after the fix above `fetchSpecificOffsets` no longer needs it,
since the expansion happens inside the callbacks, but `fetchPartitionOffsets`
still does.
##########
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:
Done. `specificOffsets` now rejects a topic carrying both forms, naming the
offending topic(s), with a test in `JsonUtilsSuite`.
This can only be reached through duplicate keys in the JSON object, which
json4s keeps as separate fields, so it used to be accepted silently and the two
entries would both end up in the limit.
--
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]
