changliiu commented on code in PR #35042:
URL: https://github.com/apache/beam/pull/35042#discussion_r2167389124
##########
sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/cache/AsyncWatermarkCache.java:
##########
@@ -42,13 +42,23 @@ public class AsyncWatermarkCache implements WatermarkCache {
private static final Object MIN_WATERMARK_KEY = new Object();
private final LoadingCache<Object, Optional<Timestamp>> cache;
+ private Timestamp lastCachedMinWatermark = Timestamp.MIN_VALUE;
+
public AsyncWatermarkCache(PartitionMetadataDao dao, Duration refreshRate) {
this.cache =
CacheBuilder.newBuilder()
.refreshAfterWrite(java.time.Duration.ofMillis(refreshRate.getMillis()))
.build(
CacheLoader.asyncReloading(
- CacheLoader.from(key ->
Optional.ofNullable(dao.getUnfinishedMinWatermark())),
+ CacheLoader.from(
+ key -> {
+ Timestamp unfinishedMinTimes =
Review Comment:
I think this is a fairly small issue.
Now:
1. the dao.getUnfinishedMinWatermarkFrom return null.
2. The CacheLoader checks do checks and return optional.ofnullable.
3. DetectNewPartitionsAction check null or not.
4. note that dao.getUnfinishedMinWatermarkFrom is also used by
NoOpWatermarkCache.
If we return optional by dao.getUnfinishedMinWatermarkFrom, then all the
above code needs to change (time-consuming and error-prone), the benefit is
only reducing one conversion from null -> optional.
Hence I lean toward keep this code as is.
I can reply so. But do let me know your thoughts on this if you disagree
then we can chat. 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.
To unsubscribe, e-mail: github-unsubscr...@beam.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
