XComp commented on code in PR #22010:
URL: https://github.com/apache/flink/pull/22010#discussion_r1497437961
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/SequenceGenerator.java:
##########
@@ -66,33 +91,47 @@ public void open(
this.checkpointedState == null,
"The " + getClass().getSimpleName() + " has already been
initialized.");
- this.checkpointedState =
- context.getOperatorStateStore()
- .getListState(
- new ListStateDescriptor<>(
- name + "-sequence-state",
LongSerializer.INSTANCE));
- this.valuesToEmit = new ArrayDeque<>();
- if (context.isRestored()) {
- // upon restoring
+ final ListStateDescriptor<SubTaskState> stateDescriptor =
+ new ListStateDescriptor<>(
+ name + "-sequence-state",
TypeInformation.of(SubTaskState.class));
+ this.checkpointedState =
context.getOperatorStateStore().getListState(stateDescriptor);
- for (Long v : this.checkpointedState.get()) {
- this.valuesToEmit.add(v);
- }
+ if (!subTaskStates.isEmpty()) {
+ this.subTaskStates.clear();
+ }
Review Comment:
```suggestion
this.subTaskStates.clear();
```
nit: clearing an empty collection shouldn't hurt.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/SequenceGenerator.java:
##########
@@ -101,26 +140,20 @@ public void snapshotState(FunctionSnapshotContext
context) throws Exception {
this.checkpointedState != null,
"The " + getClass().getSimpleName() + " state has not been
properly initialized.");
- this.checkpointedState.update(new ArrayList<>(this.valuesToEmit));
+ this.checkpointedState.clear();
+ this.checkpointedState.addAll(new ArrayList<>(subTaskStates));
Review Comment:
```suggestion
this.checkpointedState.update(new ArrayList<>(subTaskStates));
```
nit: we can continue to rely on `update` here, I guess, to keep the diff
smaller.
--
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]
