[source][flink] Add some annotated information for the benefit of interested readers #5617
Conversation
| coordinatedSource.addSplits(subtaskId, splits); | ||
| } | ||
|
|
||
| @Override | ||
| public void signalNoMoreSplits(int subtaskId) { | ||
| /* Notifies the current Task that the reader has not split to close the currently running Task */ |
There was a problem hiding this comment.
| /* Notifies the current Task that the reader has not split to close the currently running Task */ | |
| /* Notify the reader that there are no unassigned splits left */ |
…ptimize-add-source-desc
| @@ -231,6 +244,7 @@ public Map<Integer, List<byte[]>> snapshotState(long checkpointId) throws Except | |||
| StateT enumeratorState = splitEnumerator.snapshotState(checkpointId); | |||
| if (enumeratorState != null) { | |||
| byte[] enumeratorStateBytes = enumeratorStateSerializer.serialize(enumeratorState); | |||
| // -1 identification that the last status information is read | |||
There was a problem hiding this comment.
Would that be a better way to describe it ？"-1 represents the state of an pending split in SourceSplitEnumerator."
| entry.getValue().open(); | ||
| // Allocates split whose status is pending |
There was a problem hiding this comment.
Would that be a better way to describe it ？"the corresponding status is assigned to the reader"
| protected final Map<Integer, List<SplitT>> restoredSplitStateMap = new HashMap<>(); | ||
|
|
||
| protected transient volatile SourceSplitEnumerator<SplitT, StateT> splitEnumerator; | ||
| // task and Reader Map Info |
There was a problem hiding this comment.
the mapping is from subtask id to the reader info
| @@ -109,9 +117,11 @@ private void createSplitEnumerator() throws Exception { | |||
| throw new RuntimeException(e); | |||
| } | |||
| } | |||
| // The information read in the status is added to the Map | |||
There was a problem hiding this comment.
Would it be better to describe this process?
restoredState.forEach(
(subtaskId, splitBytes) -> {
if (subtaskId == -1) {
return;
}
List restoredSplitState = new ArrayList<>(splitBytes.size());
for (byte[] splitByte : splitBytes) {
try {
restoredSplitState.add(splitSerializer.deserialize(splitByte));
} catch (IOException e) {
throw new RuntimeException(e);
}
}
// The information read in the status is added to the Map
restoredSplitStateMap.put(subtaskId, restoredSplitState);
});
| coordinatedSource.addSplits(subtaskId, splits); | ||
| } | ||
|
|
||
| @Override | ||
| public void signalNoMoreSplits(int subtaskId) { | ||
| /* Notify the reader that there are no unassigned splits left */ |
There was a problem hiding this comment.
We can unify the style of interline comments and use "//" uniformly.
| protected final Map<Integer, List<SplitT>> restoredSplitStateMap = new HashMap<>(); | ||
|
|
||
| protected transient volatile SourceSplitEnumerator<SplitT, StateT> splitEnumerator; | ||
| // the mapping is from subtask id to the reader info |
There was a problem hiding this comment.
// The mapping is from subtask id to the reader info
Purpose of this pull request
Does this PR introduce any user-facing change?
How was this patch tested?
Check list
New License Guide
release-note.