dawidwys commented on a change in pull request #18702:
URL: https://github.com/apache/flink/pull/18702#discussion_r804523534
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java
##########
@@ -370,7 +411,17 @@ public DataInputStatus emitNext(DataOutput<OUT> output)
throws Exception {
private DataInputStatus emitNextNotReading(DataOutput<OUT> output) throws
Exception {
switch (operatingMode) {
case OUTPUT_NOT_INITIALIZED:
- currentMainOutput = eventTimeLogic.createMainOutput(output);
+ if (watermarkAlignmentParams.isEnabled()) {
+ // Only wrap the output when watermark alignment is
enabled, as otherwise this
+ // introduces a small performance regression (probably
because of an extra
+ // virtual call)
+ processingTimeService.scheduleWithFixedDelay(
+ this::emitLatestWatermark,
+ watermarkAlignmentParams.getUpdateInterval(),
+ watermarkAlignmentParams.getUpdateInterval());
+ }
+ currentMainOutput =
+ eventTimeLogic.createMainOutput(output,
this::onWatermarkEmitted);
Review comment:
Does it make sense to pass an empty method instead of
`onWatermarkEmitted` if watermark alignment is disabled? Right now we execute
some unnecessary logic on every watermark if the alignment is disabled.
##########
File path:
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/WatermarkToDataOutput.java
##########
@@ -35,12 +36,21 @@
public final class WatermarkToDataOutput implements WatermarkOutput {
private final PushingAsyncDataInput.DataOutput<?> output;
+ private final TimestampsAndWatermarks.OnWatermarkEmitted watermarkEmitted;
private long maxWatermarkSoFar;
private boolean isIdle;
- /** Creates a new WatermarkOutput against the given DataOutput. */
+ @VisibleForTesting
public WatermarkToDataOutput(PushingAsyncDataInput.DataOutput<?> output) {
+ this(output, watermark -> {});
+ }
+
+ /** Creates a new WatermarkOutput against the given DataOutput. */
+ public WatermarkToDataOutput(
+ PushingAsyncDataInput.DataOutput<?> output,
+ TimestampsAndWatermarks.OnWatermarkEmitted watermarkEmitted) {
this.output = checkNotNull(output);
+ this.watermarkEmitted = watermarkEmitted;
Review comment:
nit: for the sake of consistency: `checkNotNull`
--
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: issues-unsubscr...@flink.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
