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:
<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:
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]
