xuzhiwen1255 commented on code in PR #22010:
URL: https://github.com/apache/flink/pull/22010#discussion_r1261890872
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/functions/source/datagen/SequenceGenerator.java:
##########
@@ -22,29 +22,44 @@
import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
-import org.apache.flink.api.common.typeutils.base.LongSerializer;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.util.Preconditions;
+import org.apache.flink.shaded.guava30.com.google.common.collect.Queues;
+
import java.math.BigDecimal;
import java.math.MathContext;
import java.math.RoundingMode;
-import java.util.ArrayDeque;
-import java.util.Deque;
+import java.util.ArrayList;
+import java.util.NoSuchElementException;
+import java.util.Queue;
/**
* A stateful, re-scalable {@link DataGenerator} that emits each number from a
given interval
* exactly once, possibly in parallel.
+ *
+ * <p>It maintains a state internally to record the position of the current
subtask sending
+ * sequence. When the task resumes, it will continue to send the sequence
value according to the
+ * position sent by the state, until all the sequences have been sent.
+ *
+ * <p><b>IMPORTANT NOTE: </b> When the degree of parallelism increases, there
may be cases where
+ * subtasks are running empty. When the degree of parallelism decreases, there
may be cases where
+ * one subtask handles multiple states.
Review Comment:
And we use the priority queue to store state, so each subtask can be
guaranteed to be sent according to the size of the sequence.
--
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]
