codope commented on code in PR #6098:
URL: https://github.com/apache/hudi/pull/6098#discussion_r928182465
##########
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:
Yes this is not meant to work for multiple writers with multiple spark
checkpoint locations for the same table.
--
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]
