Re: [PR] [SPARK-58774][SS] Support topic-level earliest/latest values in the Kafka source startingOffsets/endingOffsets JSON [spark]
HeartSaVioR commented on PR #58043: URL: https://github.com/apache/spark/pull/58043#issuecomment-5404974792 **Merge Summary:** - merged into master https://github.com/apache/spark/commit/64b4fd2a72140e7ac43e3d974b908016f8f68148 - merged into branch-4.x https://github.com/apache/spark/commit/525f04570b0f836719c4a0dc6c00aca0a1399e6c *Posted by `merge_spark_pr.py`* -- 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]
Re: [PR] [SPARK-58774][SS] Support topic-level earliest/latest values in the Kafka source startingOffsets/endingOffsets JSON [spark]
HeartSaVioR closed pull request #58043: [SPARK-58774][SS] Support topic-level earliest/latest values in the Kafka source startingOffsets/endingOffsets JSON URL: https://github.com/apache/spark/pull/58043 -- 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]
Re: [PR] [SPARK-58774][SS] Support topic-level earliest/latest values in the Kafka source startingOffsets/endingOffsets JSON [spark]
HeartSaVioR commented on PR #58043: URL: https://github.com/apache/spark/pull/58043#issuecomment-5404965682 Thanks! Merging to master/4.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: [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]
Re: [PR] [SPARK-58774][SS] Support topic-level earliest/latest values in the Kafka source startingOffsets/endingOffsets JSON [spark]
Kino1994 commented on code in PR #58043:
URL: https://github.com/apache/spark/pull/58043#discussion_r3832333918
##
connector/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/JsonUtils.scala:
##
Review Comment:
It is still used, in three places that read offsets back from the
checkpoint: `KafkaSourceOffset`, `KafkaMicroBatchStream.deserializeOffset` and
`KafkaContinuousStream.deserializeOffset`.
I would rather keep the two separate. Persisted offsets are always fully
resolved per topic-partition, so accepting the topic-level shorthand there
would let a form the rest of the code does not expect come out of a checkpoint.
Keeping `partitionOffsets` strict is deliberate, and `specificOffsets` is only
used for the user-supplied `startingOffsets` / `endingOffsets` options. Happy
to reconsider if you would prefer them merged.
##
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:
Done, both blocks now use a single `foreach` over the per-partition and
topic-level entries keyed by name.
##
docs/streaming/structured-streaming-kafka-integration.md:
##
@@ -442,13 +451,15 @@ The following configurations are optional:
endingOffsets
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:
Done, added a mixed example to both `startingOffsets` and `endingOffsets`.
--
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]
Re: [PR] [SPARK-58774][SS] Support topic-level earliest/latest values in the Kafka source startingOffsets/endingOffsets JSON [spark]
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,
Re: [PR] [SPARK-58774][SS] Support topic-level earliest/latest values in the Kafka source startingOffsets/endingOffsets JSON [spark]
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:
endingOffsets
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
Re: [PR] [SPARK-58774][SS] Support topic-level earliest/latest values in the Kafka source startingOffsets/endingOffsets JSON [spark]
uros-b commented on PR #58043: URL: https://github.com/apache/spark/pull/58043#issuecomment-5313835998 Thank you @Kino1994! cc @HeartSaVioR who has more context on Kafka offset and metadata changes -- 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]
