Abacn commented on code in PR #39746:
URL: https://github.com/apache/beam/pull/39746#discussion_r3816221261
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Watch.java:
##########
@@ -1305,7 +1490,9 @@ public GrowthState decode(InputStream is) throws
IOException {
case NON_POLLING_GROWTH_STATE:
return decodeNonPollingGrowthState(is);
case POLLING_GROWTH_STATE:
- return decodePollingGrowthState(is);
+ return decodePollingGrowthState(is, false);
Review Comment:
This is a good design, make GrowthState coder upgrade compatible (encoded
form remains the same for existing NON_POLLING_GROWTH_STATE and
POLLING_GROWTH_STATE)
##########
sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Watch.java:
##########
@@ -899,13 +941,26 @@ public ProcessContinuation process(
return stop();
}
+ PollingGrowthState<TerminationStateT> pollingRestriction =
+ (PollingGrowthState<TerminationStateT>) currentRestriction;
+
+ Duration allowedLateness = spec.getTimestampCursorAllowedLateness();
Review Comment:
Consider `@Nullable Duration allowedLateness` as null has a meaning here.
--
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]
