danny0405 commented on code in PR #9906:
URL: https://github.com/apache/hudi/pull/9906#discussion_r1369461753
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/IncrementalInputSplits.java:
##########
@@ -407,21 +408,23 @@ private Result getHollowInputSplits(
}
@Nullable
- private InstantRange getInstantRange(String issuedInstant, String
instantToIssue, boolean nullableBoundary) {
+ private InstantRange getInstantRange(String issuedInstant, String
startInstant, String instantToIssue, boolean nullableBoundary) {
if (issuedInstant != null) {
// the streaming reader may record the last issued instant, if the
issued instant is present,
// the instant range should be: (issued instant, the latest instant].
return
InstantRange.builder().startInstant(issuedInstant).endInstant(instantToIssue)
.nullableBoundary(nullableBoundary).rangeType(InstantRange.RangeType.OPEN_CLOSE).build();
- } else if
(this.conf.getOptional(FlinkOptions.READ_START_COMMIT).isPresent()) {
- // first time consume and has a start commit
+ } else if
(this.conf.getOptional(FlinkOptions.READ_START_COMMIT).isPresent()
+ &&
!this.conf.getString(FlinkOptions.READ_START_COMMIT).equalsIgnoreCase(FlinkOptions.START_COMMIT_LATEST))
{
+ // first time consume , consumes form earliest commit or consumes from a
start commit.
final String startCommit =
this.conf.getString(FlinkOptions.READ_START_COMMIT);
return startCommit.equalsIgnoreCase(FlinkOptions.START_COMMIT_EARLIEST)
- ? null
+ ?
InstantRange.builder().startInstant(startInstant).endInstant(instantToIssue)
Review Comment:
Reading from the latest commit is the default behavior.
--
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]
