You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2021/05/20 12:38:01 UTC
[GitHub] [spark] HeartSaVioR opened a new pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
HeartSaVioR opened a new pull request #32609:
URL: https://github.com/apache/spark/pull/32609
### What changes were proposed in this pull request?
This patch is a follow-up of SPARK-26848 (#23747). In SPARK-26848, we decided to open possibility to let end users set individual timestamp per partition. But in many cases, specifying timestamp represents the intention that we would want to go back to specific timestamp and reprocess records, which should be applied to all topics and partitions.
This patch proposes to provide a way to set a global timestamp across topic-partitions which the source is subscribing to, so that end users can set all offsets by specific timestamp easily. To provide the way to config the timestamp easier, the new options only receive "a" timestamp for start/end timestamp.
New options introduced in this PR:
* startingTimestamp
* endingTimestamp
All two options receive timestamp as string.
There're priorities for options regarding starting/ending offset as we will have three options for start offsets and another three options for end offsets. Priorities are following:
* starting offsets: startingTimestamp -> startingOffsetsByTimestamp -> startingOffsets
* ending offsets: startingTimestamp -> startingOffsetsByTimestamp -> startingOffsets
### Why are the changes needed?
Existing option to specify timestamp as offset is quite verbose if there're a lot of partitions across topics. Suppose there're 100s of partitions in a topic, the json should contain 100s of times of the same timestamp.
Also, the number of partitions can also change, which requires either:
* fixing the code if the json is statically created
* introducing the dependencies on Kafka client and deal with Kafka API on crafting json programmatically
Both approaches are even not "acceptable" if we're dealing with ad-hoc query; anyone doesn't want to write the code more complicated than the query itself. Flink [provides the option](https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/datastream/kafka/#kafka-consumers-start-position-configuration) to specify a timestamp for all topic-partitions like this PR, and even doesn't provide the option to specify the timestamp per topic-partition.
### Does this PR introduce _any_ user-facing change?
Yes, this PR introduces two new options, described in above section.
### How was this patch tested?
New UTs.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845619416
**[Test build #138786 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138786/testReport)** for PR 32609 at commit [`ec1f662`](https://github.com/apache/spark/commit/ec1f662234dc3e986ae460077e16a6c557f03cf7).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
viirya commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r637761234
##########
File path: docs/structured-streaming-kafka-integration.md
##########
@@ -370,16 +384,11 @@ The following configurations are optional:
<td>none (the value of <code>startingOffsets</code> will apply)</td>
Review comment:
Change to "next preference is ..." for consistency?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r638336763
##########
File path: docs/structured-streaming-kafka-integration.md
##########
@@ -512,6 +526,17 @@ The following configurations are optional:
</tr>
</table>
+### Details on timestamp offset options
+
+The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.
+The behavior varies across options if the matched offset doesn't exist - check the description of each option.
+
+Spark simply passes the timestamp information to <code>KafkaConsumer.offsetsForTimes</code>, and doesn't interpret or reason about the value.
+For more details on <code>KafkaConsumer.offsetsForTimes</code>, please refer <a href="https://kafka.apache.org/21/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#offsetsForTimes-java.util.Map-">javadoc</a> for details.
Review comment:
If I understand correctly, there's no notion of "latest" so we picked the version we used at that time. (Worth noting that the content was added when we added timestamp offset.)
I'm OK to either raising the version to the one we use in 3.2 or lowering the version to minimum.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847868395
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43445/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847789199
**[Test build #138924 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138924/testReport)** for PR 32609 at commit [`b72f590`](https://github.com/apache/spark/commit/b72f590de83217aa30753df1a28f2b1ac9459983).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847458924
**[Test build #138897 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138897/testReport)** for PR 32609 at commit [`afd6c11`](https://github.com/apache/spark/commit/afd6c1175bd6ba9fcfd2d05b6697dc3761f1ae5c).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847704592
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43438/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] gaborgsomogyi commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r638530241
##########
File path: docs/structured-streaming-kafka-integration.md
##########
@@ -512,6 +526,17 @@ The following configurations are optional:
</tr>
</table>
+### Details on timestamp offset options
+
+The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.
+The behavior varies across options if the matched offset doesn't exist - check the description of each option.
+
+Spark simply passes the timestamp information to <code>KafkaConsumer.offsetsForTimes</code>, and doesn't interpret or reason about the value.
+For more details on <code>KafkaConsumer.offsetsForTimes</code>, please refer <a href="https://kafka.apache.org/21/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#offsetsForTimes-java.util.Map-">javadoc</a> for details.
Review comment:
I've double checked and no "latest" found. I think from maintenance perspective it would be the best to lower the version to the minimum. Such case the API must remain the same and we don't have to touch it time to time. WDYT?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-848240215
lgtm too
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847708691
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138917/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845634865
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138786/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847824760
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138924/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845671780
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43309/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] gaborgsomogyi commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r637727449
##########
File path: docs/structured-streaming-kafka-integration.md
##########
@@ -512,6 +526,17 @@ The following configurations are optional:
</tr>
</table>
+### Details on timestamp offset options
+
+The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.
+The behavior varies across options if the matched offset doesn't exist - check the description of each option.
+
+Spark simply passes the timestamp information to <code>KafkaConsumer.offsetsForTimes</code>, and doesn't interpret or reason about the value.
+For more details on <code>KafkaConsumer.offsetsForTimes</code>, please refer <a href="https://kafka.apache.org/21/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#offsetsForTimes-java.util.Map-">javadoc</a> for details.
Review comment:
Putting exact version information into the doc needs time to time attention.
`https://kafka.apache.org/21/...`
If we assume Kafka is not breaking API then we can put `latest` instead of `21`. Though not sure Kafka has such link.
BTW why `21`, we're on `<kafka.version>2.8.0</kafka.version>` and feature requires minimum `0.10.1.0`?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR closed pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR closed pull request #32609:
URL: https://github.com/apache/spark/pull/32609
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847666279
**[Test build #138917 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138917/testReport)** for PR 32609 at commit [`20b276d`](https://github.com/apache/spark/commit/20b276d2a6478576fdce73b24ec4d67371b190df).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847666279
**[Test build #138917 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138917/testReport)** for PR 32609 at commit [`20b276d`](https://github.com/apache/spark/commit/20b276d2a6478576fdce73b24ec4d67371b190df).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r638335286
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala
##########
@@ -564,20 +575,38 @@ private[kafka010] object KafkaSourceProvider extends Logging {
def getKafkaOffsetRangeLimit(
params: CaseInsensitiveMap[String],
+ globalOffsetTimestampOptionKey: String,
offsetByTimestampOptionKey: String,
offsetOptionKey: String,
defaultOffsets: KafkaOffsetRangeLimit): KafkaOffsetRangeLimit = {
- params.get(offsetByTimestampOptionKey).map(_.trim) match {
- case Some(json) => SpecificTimestampRangeLimit(JsonUtils.partitionTimestamps(json))
- case None =>
- params.get(offsetOptionKey).map(_.trim) match {
- case Some(offset) if offset.toLowerCase(Locale.ROOT) == "latest" =>
- LatestOffsetRangeLimit
- case Some(offset) if offset.toLowerCase(Locale.ROOT) == "earliest" =>
- EarliestOffsetRangeLimit
- case Some(json) => SpecificOffsetRangeLimit(JsonUtils.partitionOffsets(json))
- case None => defaultOffsets
- }
+ // The order below represents "preferences"
+
+ // 1. global timestamp
+ if (params.contains(globalOffsetTimestampOptionKey)) {
+ val tsStr = params(globalOffsetTimestampOptionKey).trim
+ try {
+ val ts = tsStr.toLong
+ return GlobalTimestampRangeLimit(ts)
Review comment:
I intended to avoid deeper level of indentations, but if ~ else if ~ else would achieve the same without return. Will address.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847708691
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138917/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] viirya commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
viirya commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-885068227
> @viirya Will it be a part of Spark 3.2.0 release?
Yes. This was merged before 3.2 branch cut.
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845619416
**[Test build #138786 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138786/testReport)** for PR 32609 at commit [`ec1f662`](https://github.com/apache/spark/commit/ec1f662234dc3e986ae460077e16a6c557f03cf7).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] rishabhsairawat commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
rishabhsairawat commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-884912145
@HeartSaVioR Thanks for this option. It makes sense to use the same timestamp across the subscribed topics.
@viirya Will it be a part of Spark 3.2.0 release?
--
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: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847749698
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43438/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847824760
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138924/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847729268
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43438/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r638387915
##########
File path: docs/structured-streaming-kafka-integration.md
##########
@@ -370,16 +384,11 @@ The following configurations are optional:
<td>none (the value of <code>startingOffsets</code> will apply)</td>
Review comment:
Nice finding!
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845634723
**[Test build #138786 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138786/testReport)** for PR 32609 at commit [`ec1f662`](https://github.com/apache/spark/commit/ec1f662234dc3e986ae460077e16a6c557f03cf7).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847443766
**[Test build #138897 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138897/testReport)** for PR 32609 at commit [`afd6c11`](https://github.com/apache/spark/commit/afd6c1175bd6ba9fcfd2d05b6697dc3761f1ae5c).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] gaborgsomogyi commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r637722144
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala
##########
@@ -564,20 +575,38 @@ private[kafka010] object KafkaSourceProvider extends Logging {
def getKafkaOffsetRangeLimit(
params: CaseInsensitiveMap[String],
+ globalOffsetTimestampOptionKey: String,
offsetByTimestampOptionKey: String,
offsetOptionKey: String,
defaultOffsets: KafkaOffsetRangeLimit): KafkaOffsetRangeLimit = {
- params.get(offsetByTimestampOptionKey).map(_.trim) match {
- case Some(json) => SpecificTimestampRangeLimit(JsonUtils.partitionTimestamps(json))
- case None =>
- params.get(offsetOptionKey).map(_.trim) match {
- case Some(offset) if offset.toLowerCase(Locale.ROOT) == "latest" =>
- LatestOffsetRangeLimit
- case Some(offset) if offset.toLowerCase(Locale.ROOT) == "earliest" =>
- EarliestOffsetRangeLimit
- case Some(json) => SpecificOffsetRangeLimit(JsonUtils.partitionOffsets(json))
- case None => defaultOffsets
- }
+ // The order below represents "preferences"
Review comment:
`specifying both global timestamp and specific timestamp for partition` added to test case 1. vs 2. fallback which is good. In order to cover all "preferences" maybe we can add a case where all config options added.
##########
File path: docs/structured-streaming-kafka-integration.md
##########
@@ -512,6 +526,17 @@ The following configurations are optional:
</tr>
</table>
+### Details on timestamp offset options
+
+The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.
+The behavior varies across options if the matched offset doesn't exist - check the description of each option.
+
+Spark simply passes the timestamp information to <code>KafkaConsumer.offsetsForTimes</code>, and doesn't interpret or reason about the value.
+For more details on <code>KafkaConsumer.offsetsForTimes</code>, please refer <a href="https://kafka.apache.org/21/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#offsetsForTimes-java.util.Map-">javadoc</a> for details.
Review comment:
Putting exact version information into the doc needs time to time attention.
`https://kafka.apache.org/21/...`
If we assume Kafka is not breaking API then we can put `latest` instead of `21`. Though not sure Kafka has such link.
BTW, why `21`, we're on `<kafka.version>2.8.0</kafka.version>` and feature requires minimum `0.10.1.0`?
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala
##########
@@ -564,20 +575,38 @@ private[kafka010] object KafkaSourceProvider extends Logging {
def getKafkaOffsetRangeLimit(
params: CaseInsensitiveMap[String],
+ globalOffsetTimestampOptionKey: String,
offsetByTimestampOptionKey: String,
offsetOptionKey: String,
defaultOffsets: KafkaOffsetRangeLimit): KafkaOffsetRangeLimit = {
- params.get(offsetByTimestampOptionKey).map(_.trim) match {
- case Some(json) => SpecificTimestampRangeLimit(JsonUtils.partitionTimestamps(json))
- case None =>
- params.get(offsetOptionKey).map(_.trim) match {
- case Some(offset) if offset.toLowerCase(Locale.ROOT) == "latest" =>
- LatestOffsetRangeLimit
- case Some(offset) if offset.toLowerCase(Locale.ROOT) == "earliest" =>
- EarliestOffsetRangeLimit
- case Some(json) => SpecificOffsetRangeLimit(JsonUtils.partitionOffsets(json))
- case None => defaultOffsets
- }
+ // The order below represents "preferences"
+
+ // 1. global timestamp
+ if (params.contains(globalOffsetTimestampOptionKey)) {
+ val tsStr = params(globalOffsetTimestampOptionKey).trim
+ try {
+ val ts = tsStr.toLong
+ return GlobalTimestampRangeLimit(ts)
Review comment:
Super nit: If we put case 2 and 3 into else then we don't need return statement.
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReader.scala
##########
@@ -77,6 +77,19 @@ private[kafka010] trait KafkaOffsetReader {
partitionTimestamps: Map[TopicPartition, Long],
failsOnNoMatchingOffset: Boolean): KafkaSourceOffset
+ /**
+ * Resolves the specific offsets based on timestamp per all topic-partitions being subscribed.
+ * The returned offset for each partition is the earliest offset whose timestamp is greater
+ * than or equal to the given timestamp in the corresponding partition. If the matched offset
Review comment:
Nit: `If the matched offset doesn't exist` is a bit odd construct. Either matched or doesn't exist.
##########
File path: docs/structured-streaming-kafka-integration.md
##########
@@ -512,6 +526,17 @@ The following configurations are optional:
</tr>
</table>
+### Details on timestamp offset options
+
+The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.
+The behavior varies across options if the matched offset doesn't exist - check the description of each option.
Review comment:
Same as down below.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847461347
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845149564
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43278/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847461343
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847460991
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43418/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r638374368
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaSourceProvider.scala
##########
@@ -564,20 +575,38 @@ private[kafka010] object KafkaSourceProvider extends Logging {
def getKafkaOffsetRangeLimit(
params: CaseInsensitiveMap[String],
+ globalOffsetTimestampOptionKey: String,
offsetByTimestampOptionKey: String,
offsetOptionKey: String,
defaultOffsets: KafkaOffsetRangeLimit): KafkaOffsetRangeLimit = {
- params.get(offsetByTimestampOptionKey).map(_.trim) match {
- case Some(json) => SpecificTimestampRangeLimit(JsonUtils.partitionTimestamps(json))
- case None =>
- params.get(offsetOptionKey).map(_.trim) match {
- case Some(offset) if offset.toLowerCase(Locale.ROOT) == "latest" =>
- LatestOffsetRangeLimit
- case Some(offset) if offset.toLowerCase(Locale.ROOT) == "earliest" =>
- EarliestOffsetRangeLimit
- case Some(json) => SpecificOffsetRangeLimit(JsonUtils.partitionOffsets(json))
- case None => defaultOffsets
- }
+ // The order below represents "preferences"
Review comment:
Yeah I changed the test a bit to cover two cases: all options / timestamp per partition vs offset. Thanks for the suggestion.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r638338036
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReader.scala
##########
@@ -77,6 +77,19 @@ private[kafka010] trait KafkaOffsetReader {
partitionTimestamps: Map[TopicPartition, Long],
failsOnNoMatchingOffset: Boolean): KafkaSourceOffset
+ /**
+ * Resolves the specific offsets based on timestamp per all topic-partitions being subscribed.
+ * The returned offset for each partition is the earliest offset whose timestamp is greater
+ * than or equal to the given timestamp in the corresponding partition. If the matched offset
Review comment:
Let me refine the words. Basically it queries to Kafka and we are explaining the mechanism, it seems OK to say `If Kafka doesn't return the matched offset`.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845149564
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43278/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847815128
**[Test build #138924 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138924/testReport)** for PR 32609 at commit [`b72f590`](https://github.com/apache/spark/commit/b72f590de83217aa30753df1a28f2b1ac9459983).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847868395
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43445/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] gaborgsomogyi commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847627288
When the hardcoded version discussion is resolved then it's good to go from my perspective.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] gaborgsomogyi commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
gaborgsomogyi commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r637726496
##########
File path: docs/structured-streaming-kafka-integration.md
##########
@@ -512,6 +526,17 @@ The following configurations are optional:
</tr>
</table>
+### Details on timestamp offset options
+
+The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.
+The behavior varies across options if the matched offset doesn't exist - check the description of each option.
Review comment:
Same as down below in the file.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845091923
**[Test build #138756 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138756/testReport)** for PR 32609 at commit [`f8da576`](https://github.com/apache/spark/commit/f8da5763e76d77171f1169fba58710b42251b4b4).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845690788
cc. @viirya @gaborgsomogyi @xuanyuanking
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845707438
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43309/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] xuanyuanking commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
xuanyuanking commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r638624224
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReader.scala
##########
@@ -77,6 +77,19 @@ private[kafka010] trait KafkaOffsetReader {
partitionTimestamps: Map[TopicPartition, Long],
failsOnNoMatchingOffset: Boolean): KafkaSourceOffset
+ /**
+ * Resolves the specific offsets based on timestamp per all topic-partitions being subscribed.
+ * The returned offset for each partition is the earliest offset whose timestamp is greater
+ * than or equal to the given timestamp in the corresponding partition. If the matched offset
+ * doesn't exist, depending on `failsOnNoMatchingOffset` parameter, the offset will be set to
+ * latest or this method throws an error.
+ *
+ * @param timestamp the timestamp.
+ * @param failsOnNoMatchingOffset whether to fail the query when no matched offset can be found.
+ */
+ def fetchGlobalTimestampBasedOffsets(timestamp: Long,
Review comment:
super nit for the code style, maybe a new line for the params?
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845149525
Kubernetes integration test unable to build dist.
exiting with code: 1
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43278/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847699925
**[Test build #138917 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138917/testReport)** for PR 32609 at commit [`20b276d`](https://github.com/apache/spark/commit/20b276d2a6478576fdce73b24ec4d67371b190df).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847821319
Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43445/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847749698
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43438/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845634865
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138786/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845688692
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43309/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845139904
**[Test build #138756 has finished](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138756/testReport)** for PR 32609 at commit [`f8da576`](https://github.com/apache/spark/commit/f8da5763e76d77171f1169fba58710b42251b4b4).
* This patch passes all tests.
* This patch merges cleanly.
* This patch adds no public classes.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r638690558
##########
File path: external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaOffsetReader.scala
##########
@@ -77,6 +77,19 @@ private[kafka010] trait KafkaOffsetReader {
partitionTimestamps: Map[TopicPartition, Long],
failsOnNoMatchingOffset: Boolean): KafkaSourceOffset
+ /**
+ * Resolves the specific offsets based on timestamp per all topic-partitions being subscribed.
+ * The returned offset for each partition is the earliest offset whose timestamp is greater
+ * than or equal to the given timestamp in the corresponding partition. If the matched offset
+ * doesn't exist, depending on `failsOnNoMatchingOffset` parameter, the offset will be set to
+ * latest or this method throws an error.
+ *
+ * @param timestamp the timestamp.
+ * @param failsOnNoMatchingOffset whether to fail the query when no matched offset can be found.
+ */
+ def fetchGlobalTimestampBasedOffsets(timestamp: Long,
Review comment:
Nice finding! Fixed it.
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847836594
I'm merging this as I got approvals with comments for a few nits, and I addressed all of them. Thanks all for the reviewing!
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HeartSaVioR commented on a change in pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
HeartSaVioR commented on a change in pull request #32609:
URL: https://github.com/apache/spark/pull/32609#discussion_r638541266
##########
File path: docs/structured-streaming-kafka-integration.md
##########
@@ -512,6 +526,17 @@ The following configurations are optional:
</tr>
</table>
+### Details on timestamp offset options
+
+The returned offset for each partition is the earliest offset whose timestamp is greater than or equal to the given timestamp in the corresponding partition.
+The behavior varies across options if the matched offset doesn't exist - check the description of each option.
+
+Spark simply passes the timestamp information to <code>KafkaConsumer.offsetsForTimes</code>, and doesn't interpret or reason about the value.
+For more details on <code>KafkaConsumer.offsetsForTimes</code>, please refer <a href="https://kafka.apache.org/21/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html#offsetsForTimes-java.util.Map-">javadoc</a> for details.
Review comment:
That makes sense. Hopefully Kafka community looks to maintain all versions of doc so the risk of broken link is relatively low. I'll update the link to 0.10.1.x. Thanks!
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847443766
**[Test build #138897 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138897/testReport)** for PR 32609 at commit [`afd6c11`](https://github.com/apache/spark/commit/afd6c1175bd6ba9fcfd2d05b6697dc3761f1ae5c).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845091923
**[Test build #138756 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138756/testReport)** for PR 32609 at commit [`f8da576`](https://github.com/apache/spark/commit/f8da5763e76d77171f1169fba58710b42251b4b4).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847789199
**[Test build #138924 has started](https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder/138924/testReport)** for PR 32609 at commit [`b72f590`](https://github.com/apache/spark/commit/b72f590de83217aa30753df1a28f2b1ac9459983).
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845144922
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138756/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] SparkQA commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
SparkQA commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-847843059
Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/43445/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins commented on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins commented on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845144922
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder/138756/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] AmplabJenkins removed a comment on pull request #32609: [SPARK-29223][SQL][SS] New option to specify timestamp on all subscribing topic-partitions in Kafka source
Posted by GitBox <gi...@apache.org>.
AmplabJenkins removed a comment on pull request #32609:
URL: https://github.com/apache/spark/pull/32609#issuecomment-845707438
Refer to this link for build results (access rights to CI server needed):
https://amplab.cs.berkeley.edu/jenkins//job/SparkPullRequestBuilder-K8s/43309/
--
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org