Github user aljoscha commented on a diff in the pull request:
https://github.com/apache/flink/pull/1895#discussion_r61445422
--- Diff:
flink-core/src/main/java/org/apache/flink/api/common/io/DelimitedInputFormat.java
---
@@ -427,18 +434,37 @@ public void open(FileInputSplit split) throws
IOException {
this.overLimit = false;
this.end = false;
- if (this.splitStart != 0) {
+ if (this.splitStart != 0 && this.restoredOffset == null) {
this.stream.seek(this.splitStart);
+ this.offset = this.splitStart;
readLine();
-
+
// if the first partial record already pushes the
stream over the limit of our split, then no
- // record starts within this split
+ // record starts within this split
if (this.overLimit) {
this.end = true;
}
+ } else if (this.restoredOffset != null) {
+
+ if (!this.restoredSplit.equals(split)) {
+ throw new RuntimeException("Tried to open at
the wrong split after recovery.");
+ }
+
+ // this is the case where we restart from a specific
offset within a split (e.g. after a node failure)
+ this.stream.seek(this.restoredOffset);
+ this.currSplit = this.restoredSplit;
+ this.offset = this.restoredOffset;
+ this.splitLength = this.splitStart + this.splitLength -
this.offset;
--- End diff --
`splitLength` is used to keep track of how many bytes there are left to
read, correct? The name is not very good in that case, which is not your fault,
since it was there before.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
