xushiyan commented on code in PR #6098:
URL: https://github.com/apache/hudi/pull/6098#discussion_r928172917
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/hudi/HoodieStreamingSink.scala:
##########
@@ -247,4 +285,18 @@ class HoodieStreamingSink(sqlContext: SQLContext,
writeClient = Option.empty
}
}
+
+ private def canSkipBatch(batchId: Long): Boolean = {
+ // get the latest checkpoint from the commit metadata to check if the
microbatch has already been prcessed or not
+ val lastCommit =
metaClient.get.getActiveTimeline.getCommitsTimeline.filterCompletedInstants().lastInstant()
+ if (lastCommit.isPresent) {
+ val commitMetadata = HoodieCommitMetadata.fromBytes(
+
metaClient.get.getActiveTimeline.getInstantDetails(lastCommit.get()).get(),
classOf[HoodieCommitMetadata])
+ val lastCheckpoint = commitMetadata.getMetadata(SinkCheckpointKey)
+ if (!StringUtils.isNullOrEmpty(lastCheckpoint)) {
+ latestBatchId = lastCheckpoint.toLong
+ }
+ }
+ latestBatchId >= batchId
Review Comment:
so we deserialize the batch id from storage. how can we guarantee is always
monotonically increasing? users can start a completely new streaming job to
the same table, then will the ids become not comparable?
--
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]
