zhijiangW commented on a change in pull request #9717: [FLINK-14044] [runtime]
Reducing synchronization in AsyncWaitOperator
URL: https://github.com/apache/flink/pull/9717#discussion_r327962005
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/async/queue/StreamRecordQueueEntry.java
##########
@@ -21,65 +21,55 @@
import org.apache.flink.annotation.Internal;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
+import org.apache.flink.streaming.api.operators.TimestampedCollector;
import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.util.Preconditions;
+
+import javax.annotation.Nonnull;
import java.util.Collection;
-import java.util.concurrent.CompletableFuture;
/**
* {@link StreamElementQueueEntry} implementation for {@link StreamRecord}.
This class also acts
* as the {@link ResultFuture} implementation which is given to the {@link
AsyncFunction}. The
* async function completes this class with a collection of results.
*
- * @param <OUT> Type of the asynchronous collection result
+ * @param <OUT> Type of the asynchronous collection result.
*/
@Internal
-public class StreamRecordQueueEntry<OUT> extends
StreamElementQueueEntry<Collection<OUT>>
- implements AsyncCollectionResult<OUT>, ResultFuture<OUT> {
-
- /** Timestamp information. */
- private final boolean hasTimestamp;
- private final long timestamp;
-
- /** Future containing the collection result. */
- private final CompletableFuture<Collection<OUT>> resultFuture;
+class StreamRecordQueueEntry<OUT> implements StreamElementQueueEntry<OUT> {
+ private Collection<OUT> completed;
- public StreamRecordQueueEntry(StreamRecord<?> streamRecord) {
- super(streamRecord);
+ @Nonnull
+ private final StreamRecord<?> inputRecord;
- hasTimestamp = streamRecord.hasTimestamp();
- timestamp = streamRecord.getTimestamp();
-
- resultFuture = new CompletableFuture<>();
+ StreamRecordQueueEntry(@Nonnull StreamRecord<?> inputRecord) {
+ this.inputRecord = Preconditions.checkNotNull(inputRecord);
}
@Override
- public boolean hasTimestamp() {
- return hasTimestamp;
+ public boolean isDone() {
+ return completed != null;
}
+ @Nonnull
@Override
- public long getTimestamp() {
- return timestamp;
+ public StreamRecord<?> getInputElement() {
+ return inputRecord;
}
@Override
- public Collection<OUT> get() throws Exception {
- return resultFuture.get();
- }
+ public void emitResult(TimestampedCollector<OUT> output) {
+ Preconditions.checkState(completed != null, "Not done yet");
- @Override
- protected CompletableFuture<Collection<OUT>> getFuture() {
- return resultFuture;
+ output.setTimestamp(this.inputRecord);
Review comment:
Remove `this`
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
