You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Sxnan (via GitHub)" <gi...@apache.org> on 2023/10/13 07:40:29 UTC
[PR] [FLINK-33202][runtime] Support switching from batch to stream mode to improve throughput when processing backlog data [flink]
Sxnan opened a new pull request, #23521:
URL: https://github.com/apache/flink/pull/23521
<!--
*Thank you very much for contributing to Apache Flink - we are happy that you want to help us improve Flink. To help the community review your contribution in the best possible way, please go through the checklist below, which will get the contribution into a shape in which it can be best reviewed.*
*Please understand that we do not do this to make contributions to Flink a hassle. In order to uphold a high standard of quality for code contributions, while at the same time managing a large number of contributions, we need contributors to prepare the contributions well, and give reviewers enough contextual information for the review. Please also understand that contributions that do not follow this guide will take longer to review and thus typically be picked up with lower priority by the community.*
## Contribution Checklist
- Make sure that the pull request corresponds to a [JIRA issue](https://issues.apache.org/jira/projects/FLINK/issues). Exceptions are made for typos in JavaDoc or documentation files, which need no JIRA issue.
- Name the pull request in the form "[FLINK-XXXX] [component] Title of the pull request", where *FLINK-XXXX* should be replaced by the actual issue number. Skip *component* if you are unsure about which is the best component.
Typo fixes that have no associated JIRA issue should be named following this pattern: `[hotfix] [docs] Fix typo in event time introduction` or `[hotfix] [javadocs] Expand JavaDoc for PuncuatedWatermarkGenerator`.
- Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review.
- Make sure that the change passes the automated tests, i.e., `mvn clean verify` passes. You can set up Azure Pipelines CI to do that following [this guide](https://cwiki.apache.org/confluence/display/FLINK/Azure+Pipelines#AzurePipelines-Tutorial:SettingupAzurePipelinesforaforkoftheFlinkrepository).
- Each pull request should address only one issue, not mix up code from multiple issues.
- Each commit in the pull request has a meaningful commit message (including the JIRA id)
- Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below.
**(The sections below can be removed for hotfixes of typos)**
-->
## What is the purpose of the change
## Brief change log
*(for example:)*
- *The TaskInfo is stored in the blob store on job creation time as a persistent artifact*
- *Deployments RPC transmits only the blob storage reference*
- *TaskManagers retrieve the TaskInfo from the blob cache*
## Verifying this change
Please make sure both new and modified tests in this PR follows the conventions defined in our code quality guide: https://flink.apache.org/contributing/code-style-and-quality-common.html#testing
*(Please pick either of the following options)*
This change is a trivial rework / code cleanup without any test coverage.
*(or)*
This change is already covered by existing tests, such as *(please describe tests)*.
*(or)*
This change added tests and can be verified as follows:
*(example:)*
- *Added integration tests for end-to-end deployment with large payloads (100MB)*
- *Extended integration test for recovery after master (JobManager) failure*
- *Added test that validates that TaskInfo is transferred only once across recoveries*
- *Manually verified the change by running a 4 node cluster with 2 JobManagers and 4 TaskManagers, a stateful streaming program, and killing one JobManager and two TaskManagers during the execution, verifying that recovery happens correctly.*
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (yes / no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (yes / no)
- The serializers: (yes / no / don't know)
- The runtime per-record code paths (performance sensitive): (yes / no / don't know)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (yes / no / don't know)
- The S3 file system connector: (yes / no / don't know)
## Documentation
- Does this pull request introduce a new feature? (yes / no)
- If yes, how is the feature documented? (not applicable / docs / JavaDocs / not documented)
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "mlegore (via GitHub)" <gi...@apache.org>.
mlegore commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-2852375243
What is the status of this work?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1384472347
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -569,6 +571,12 @@ public void handleOperatorEvent(OperatorEvent event) {
sourceReader.handleSourceEvents(((SourceEventWrapper) event).getSourceEvent());
} else if (event instanceof NoMoreSplitsEvent) {
sourceReader.notifyNoMoreSplits();
+ } else if (event instanceof IsBacklogEvent) {
+ eventTimeLogic.triggerPeriodicEmit(System.currentTimeMillis());
Review Comment:
We should invoke this method to send the watermark downstream in both cases. In case of event.isBacklog() == false, the downstream need to update the max watermark during backlog.
We cannot cancel the periodicEmit. Otherwise, operators that do not support backlog processing, such as multi-input operators, will not work.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #23521: [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator
URL: https://github.com/apache/flink/pull/23521
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "SmirAlex (via GitHub)" <gi...@apache.org>.
SmirAlex commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-1814359886
Thanks for the response, @Sxnan. I got your points. BTW, the reason why delivery guarantees of `RecordAttributes` elements to downstream operators was important to me is that I was testing the possibility to use them in order to properly implement Processing Time Temporal Join. I [described](https://lists.apache.org/thread/b3doy1gqchv45g2vjxcblv10two9gvsv) my thoughts about this in [FLIP-326](https://cwiki.apache.org/confluence/display/FLINK/FLIP-326%3A+Enhance+Watermark+to+Support+Processing-Time+Temporal+Join) discussion thread. Thus, if eventually it will be decided to reuse `RecordAttributes` logic to solve [FLINK-19830](https://issues.apache.org/jira/browse/FLINK-19830), the problem I described will arise one way or another.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-1798020804
@yunfengzhou-hub Thanks for the review! I updated the PR. Can you have another look?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "SmirAlex (via GitHub)" <gi...@apache.org>.
SmirAlex commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-1812134103
Hi @Sxnan! Sorry, I understand that I'm not a reviewer, but it happened that I was testing functionality from this MR recently, and I found a bug (in my opinion). It concerns the logic of `RecordAttributesValve`.
There was very little backlog data in my test at source + it had parallelism 4 (actually, any value > 1 suits the case). Due to very short backlog phase, time interval between sending `RecordAttributes(isBacklog=true)` and `RecordAttributes(isBacklog=false)` was also very short. In addition, due to the high parallelism one source subtask could send `RecordAttributes(isBacklog=false)` even before `RecordAttributes(isBacklog=true) `of another subtask. As a result, race condition have occurred in `RecordAttributesValve#inputRecordAttributes`. `backlogChannelsCnt` was incrementing and decrementing simultaneously, which led to not reaching `numInputChannels`, so no RecordAttributes was emitted from RecordAttributesValve at all.
I suggest to have different counters for `RecordAttributes(isBacklog=true)` and for `RecordAttributes(isBacklog=false)`. Therefore, the race condition I mentioned earlier won't affect the result. Something like this:
```
if (recordAttributes.isBacklog()) {
backlogChannelsCnt += 1;
if (backlogChannelsCnt != numInputChannels) {
return;
}
backlogChannelsCnt = 0;
} else {
nonBacklogChannelsCnt += 1;
if (nonBacklogChannelsCnt != numInputChannels) {
return;
}
nonBacklogChannelsCnt = 0;
}
if (lastOutputAttributes == null
|| lastOutputAttributes.isBacklog() != recordAttributes.isBacklog()) {
if (lastOutputAttributes != null && !lastOutputAttributes.isBacklog()) {
LOG.warn(
"Switching from non-backlog to backlog is currently not supported. Backlog status remains.");
return;
}
lastOutputAttributes = recordAttributes;
output.emitRecordAttributes(recordAttributes);
}
```
WDYT? Also, it is good to have a test for aforementioned case.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33202][runtime] Support switching from batch to stream mode to improve throughput when processing backlog data [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1369927150
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/TestTriggerable.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.assertj.core.api.Assertions;
+
+import java.util.function.Consumer;
+
+/** Triggerable for test. */
+class TestTriggerable<K, N> implements Triggerable<K, N> {
+
+ private final Consumer<InternalTimer<K, N>> eventTimeHandler;
+ private final Consumer<InternalTimer<K, N>> processingTimeHandler;
+
+ public static <K, N> TestTriggerable<K, N> eventTimeTrigger(
+ Consumer<InternalTimer<K, N>> eventTimeHandler) {
+ return new TestTriggerable<>(
+ eventTimeHandler,
+ timer -> Assertions.fail("We did not expect processing timer to be triggered."));
+ }
+
+ public static <K, N> TestTriggerable<K, N> processingTimeTrigger(
Review Comment:
I updated the test `BatchExecutionInternalTimeServiceTest` so that the method is used.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-1853204464
After an offline discussion with @xintongsong, we will split this PR into two PRs. This PR will focus on optimizing the one input operator during backlog. The other [PR](https://github.com/apache/flink/pull/23919) will focus on propagating the RecordAttributes through the job graph.
This PR should be merged after the other PR.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33202][runtime] Support switching from batch to stream mode to improve throughput when processing backlog data [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1366607532
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java:
##########
@@ -415,6 +417,12 @@ private void configureStreamGraphStreaming(final StreamGraph graph) {
graph.setCheckpointStorage(checkpointStorage);
graph.setSavepointDirectory(savepointDir);
graph.setGlobalStreamExchangeMode(deriveGlobalStreamExchangeModeStreaming());
+
+ if (Duration.ZERO.equals(
+ configuration.get(
+ ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL_DURING_BACKLOG))) {
+ graph.setTimerServiceProvider(InternalBacklogAwareTimerServiceManagerImpl::create);
+ }
Review Comment:
Before this PR, if it is not set, it will use the `InternalTimeServiceManagerImpl::create` as the timerSerivceProvider, see `org.apache.flink.streaming.api.operators.InternalTimeServiceManagerImpl#705`
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33202][runtime] Support switching from batch to stream mode to improve throughput when processing backlog data [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1366593279
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java:
##########
@@ -649,4 +659,52 @@ public OperatorID getOperatorID() {
protected Optional<InternalTimeServiceManager<?>> getTimeServiceManager() {
return Optional.ofNullable(timeServiceManager);
}
+
+ @Experimental
+ public void processRecordAttributes(RecordAttributes recordAttributes) throws Exception {
+ lastRecordAttributes1 = recordAttributes;
+ if (timeServiceManager != null
+ && timeServiceManager instanceof InternalBacklogAwareTimerServiceManagerImpl) {
+ final InternalBacklogAwareTimerServiceManagerImpl<?> backlogAwareTimerServiceManager =
+ (InternalBacklogAwareTimerServiceManagerImpl<?>) timeServiceManager;
+ if (recordAttributes instanceof InternalRecordAttributes) {
+ backlogAwareTimerServiceManager.setMaxWatermarkDuringBacklog(
+ ((InternalRecordAttributes) recordAttributes)
+ .getMaxWatermarkDuringBacklog());
+ }
+ backlogAwareTimerServiceManager.setBacklog(recordAttributes.isBacklog());
+ }
+ output.emitRecordAttributes(
+ new RecordAttributesBuilder(Collections.singletonList(recordAttributes)).build());
+ }
+
+ @Experimental
+ public void processRecordAttributes1(RecordAttributes recordAttributes) throws Exception {
+ lastRecordAttributes1 = recordAttributes;
+ List<RecordAttributes> lastRecordAttributes = getTwoInputsLastRecordAttributes();
+ output.emitRecordAttributes(new RecordAttributesBuilder(lastRecordAttributes).build());
+ }
+
+ @Experimental
+ public void processRecordAttributes2(RecordAttributes recordAttributes) throws Exception {
+ lastRecordAttributes2 = recordAttributes;
+ List<RecordAttributes> lastRecordAttributes = getTwoInputsLastRecordAttributes();
+ output.emitRecordAttributes(new RecordAttributesBuilder(lastRecordAttributes).build());
+ }
+
+ private List<RecordAttributes> getTwoInputsLastRecordAttributes() {
+ List<RecordAttributes> lastRecordAttributes;
+ if (lastRecordAttributes1 == null && lastRecordAttributes2 == null) {
+ // should not reach here.
+ throw new RuntimeException(
+ "lastRecordAttributes1 and lastRecordAttributes2 cannot be both null.");
Review Comment:
I think it makes more sense to make the precondition check with Preconditions#checkState.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-2777666770
This PR has been closed since it has not had any activity in 120 days.
If you feel like this was a mistake, or you would like to continue working on it,
please feel free to re-open the PR and ask for a review.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33202][runtime] Support switching from batch to stream mode to improve throughput when processing backlog data [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-1765550224
@flinkbot run azure
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-1813700885
> Hi @Sxnan! Sorry, I understand that I'm not a reviewer, but it happened that I was testing functionality from this MR recently, and I found a bug (in my opinion). It concerns the logic of `RecordAttributesValve`. There was very little backlog data in my test at source + it had parallelism 4 (actually, any value > 1 suits the case). Due to very short backlog phase, time interval between sending `RecordAttributes(isBacklog=true)` and `RecordAttributes(isBacklog=false)` was also very short. In addition, due to the high parallelism one source subtask could send `RecordAttributes(isBacklog=false)` even before `RecordAttributes(isBacklog=true) `of another subtask. As a result, race condition have occurred in `RecordAttributesValve#inputRecordAttributes`. `backlogChannelsCnt` was incrementing and decrementing simultaneously, which led to not reaching `numInputChannels`, so no RecordAttributes was emitted from RecordAttributesValve at all.
>
> I suggest to have different counters for `RecordAttributes(isBacklog=true)` and for `RecordAttributes(isBacklog=false)`. Therefore, the race condition I mentioned earlier won't affect the result. Something like this:
>
> ```
> if (recordAttributes.isBacklog()) {
> backlogChannelsCnt += 1;
> if (backlogChannelsCnt != numInputChannels) {
> return;
> }
> backlogChannelsCnt = 0;
> } else {
> nonBacklogChannelsCnt += 1;
> if (nonBacklogChannelsCnt != numInputChannels) {
> return;
> }
> nonBacklogChannelsCnt = 0;
> }
>
> if (lastOutputAttributes == null
> || lastOutputAttributes.isBacklog() != recordAttributes.isBacklog()) {
> if (lastOutputAttributes != null && !lastOutputAttributes.isBacklog()) {
> LOG.warn(
> "Switching from non-backlog to backlog is currently not supported. Backlog status remains.");
> return;
> }
> lastOutputAttributes = recordAttributes;
> output.emitRecordAttributes(recordAttributes);
> }
> ```
>
> WDYT? Also, it is good to have a test for aforementioned case.
Hi @Smir. Thanks for trying this out! The RecordAttributesValve combines the RecodAttributes from different input channels from the same input. The input is considered in backlog state if and only if all the input channels are backlog = true. Otherwise, some non-backlog records will be considered as backlog records.
Back to your case, where there is very little backlog data and it has parallelism greater than 1. It is possible that an input channel switches to non-backlog state before other input channels switch to backlog state. When that happens, we just process those data as if they are non-backlog data.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-1809430382
@xintongsong Could you help review this PR?
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "SmirAlex (via GitHub)" <gi...@apache.org>.
SmirAlex commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-1814413557
Thanks for the response, @Sxnan. I got your points. BTW, the reason why delivery guarantees of `RecordAttributes` elements to downstream operators was important to me is that I was testing the possibility to use them in order to properly implement Processing Time Temporal Join. I [described](https://lists.apache.org/thread/b3doy1gqchv45g2vjxcblv10two9gvsv) my thoughts about this in [FLIP-326](https://cwiki.apache.org/confluence/display/FLINK/FLIP-326%3A+Enhance+Watermark+to+Support+Processing-Time+Temporal+Join) discussion thread. Thus, if eventually it will be decided to reuse `RecordAttributes` logic in order to solve [FLINK-19830](https://issues.apache.org/jira/browse/FLINK-19830), the problem I described will arise one way or another.
Also, I should mention that there is another workaround that guarantees delivery of `RecordAttributes(isBacklog=false)` (but not isBacklog=true), but at the same time prevents considering some non-backlog records as backlog records. Pseudocode may be like this:
```
if (lastOutputAttributes == null && nonBacklogChannelsCnt == numInputChannels) {
nonBacklogChannelsCnt = 0;
output.emitRecordAttributes(RecordAttributes(isBacklog=false));
}
```
The downside is that it is possible to have `RecordAttributes(isBacklog=false)` without `RecordAttributes(isBacklog=true)` previously, but probably, it's ok. In other respects, this approach worked fine for me.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "xintongsong (via GitHub)" <gi...@apache.org>.
xintongsong commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1406042500
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/RecordAttributes.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.streamrecord;
+
+import org.apache.flink.annotation.Experimental;
+
+import java.util.Objects;
+
+/**
+ * A RecordAttributes element provides stream task with information that can be used to optimize the
+ * stream task's performance.
+ */
+@Experimental
+public class RecordAttributes extends StreamElement {
+ private final boolean backlog;
Review Comment:
Minor: I'd suggest the name `isBacklog`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/source/event/BacklogEvent.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.source.event;
+
+import org.apache.flink.runtime.operators.coordination.OperatorEvent;
+
+import java.util.Objects;
+
+/** A source event that notify the source operator of the backlog status. */
+public class BacklogEvent implements OperatorEvent {
+
+ private static final long serialVersionUID = 1L;
+ private final boolean backlog;
+
+ public BacklogEvent(boolean backlog) {
+ this.backlog = backlog;
+ }
+
+ public boolean isBacklog() {
+ return backlog;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ final BacklogEvent that = (BacklogEvent) o;
+ return Objects.equals(backlog, that.backlog);
+ }
Review Comment:
It would be a good practice to always implement the `hashCode()` together with `equals`, in case the class is accidentally used for a hash-map/set.
##########
flink-runtime/src/test/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContextTest.java:
##########
@@ -261,4 +262,26 @@ private List<ReaderInfo> registerReaders() {
return infos;
}
+
+ @Test
+ void testSetIsProcessingBacklog() throws Exception {
+ sourceReady();
+ registerReader(0, 0);
+ context.setIsProcessingBacklog(true);
+
+ Thread.sleep(10); // sleep a little for the coordinator to handle reader register event.
Review Comment:
Such sleep-based approach is usually problematic. If the time is too short, this could become a source of instability. If the time is too long, it slows down the entire CI.
I'd suggest to keep checking whether the expected condition is reached with a short interval, and to fail if the condition is not reached within a relative long timeout.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/RecordAttributes.java:
##########
@@ -0,0 +1,66 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.streamrecord;
+
+import org.apache.flink.annotation.Experimental;
+
+import java.util.Objects;
+
+/**
+ * A RecordAttributes element provides stream task with information that can be used to optimize the
+ * stream task's performance.
+ */
Review Comment:
The semantic here is a bit vague. Given the name `RecordAttributes`, this class should provide information about some records. The question is which records it describes? IIUC, the information for RA_i should apply to all records received between RA_i and RA_i+1. This should be explicitly explained as part of the class JavaDoc.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/source/event/BacklogEvent.java:
##########
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.source.event;
+
+import org.apache.flink.runtime.operators.coordination.OperatorEvent;
+
+import java.util.Objects;
+
+/** A source event that notify the source operator of the backlog status. */
+public class BacklogEvent implements OperatorEvent {
+
+ private static final long serialVersionUID = 1L;
+ private final boolean backlog;
Review Comment:
Minor: I'd name this field (and corresponding method arguments) `isProcessingBacklog`.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -569,6 +571,14 @@ public void handleOperatorEvent(OperatorEvent event) {
sourceReader.handleSourceEvents(((SourceEventWrapper) event).getSourceEvent());
} else if (event instanceof NoMoreSplitsEvent) {
sourceReader.notifyNoMoreSplits();
+ } else if (event instanceof BacklogEvent) {
+ if (eventTimeLogic != null) {
+ eventTimeLogic.triggerPeriodicEmit(System.currentTimeMillis());
+ }
Review Comment:
IIUC, this is to always trigger an emit of watermark upon changing of `isBacklog`? I think it was fine with the method `triggerPeriodicEmit` when the method is private to `ProgressiveTimestampsAndWatermarks` and is only used for `scheduleWithFixedDelay`. However, when it turns into a public method, the name becomes confusing. How could a PERIODIC thing being manually triggered in a non-periodical way? It might make sense to simply rename this method to something like `emitImmediateWatermark`.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/Output.java:
##########
@@ -55,4 +57,13 @@ public interface Output<T> extends Collector<T> {
<X> void collect(OutputTag<X> outputTag, StreamRecord<X> record);
void emitLatencyMarker(LatencyMarker latencyMarker);
+
+ /**
+ * Emits a {@link RecordAttributes} from an operator. This element is broadcast to all
+ * downstream operators.
+ */
+ @Experimental
+ default void emitRecordAttributes(RecordAttributes recordAttributes) {
+ throw new UnsupportedOperationException();
+ }
Review Comment:
I think an `Output` is meant to be called by users, but should never be implemented by users. You might need to double-check on this, because I'm not 100% sure.
I'm bringing this up because, if this is never implemented by users, we don't need to provide such a default implementation that only throws an error. Otherwise, if there are chances that this is implemented by users, then this is a breaking change because the method can be called by framework and thus the error will be thrown anyway.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/SortingDataInput.java:
##########
@@ -183,6 +184,9 @@ public void emitWatermarkStatus(WatermarkStatus watermarkStatus) {}
@Override
public void emitLatencyMarker(LatencyMarker latencyMarker) {}
+
+ @Override
+ public void emitRecordAttributes(RecordAttributes recordAttributes) {}
Review Comment:
Just trying to understand. Is it possible that a `SortingDataInput` or a `MultiInputSortingDataInput` receive a `RecordAttributes`? If yes, what happens given that we are not forwarding it? If not, we may consider adding some sanity checks.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/CollectionDataInput.java:
##########
@@ -53,6 +53,10 @@ public DataInputStatus emitNext(DataOutput<E> output) throws Exception {
output.emitRecord(streamElement.asRecord());
} else if (streamElement instanceof Watermark) {
output.emitWatermark(streamElement.asWatermark());
+ } else if (streamElement.isRecordAttributes()) {
+ output.emitRecordAttributes(streamElement.asRecordAttributes());
+ } else if (streamElement.isWatermarkStatus()) {
+ output.emitWatermarkStatus(streamElement.asWatermarkStatus());
Review Comment:
How does this relate to this PR? Is this a fix of a previous bug? If yes, this should be placed in a separate hotfix commit before other commits.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/RecordAttributesValve.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput.DataOutput;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+
+/**
+ * RecordAttributesValve combine RecordAttributes from different input channels. If any of the input
+ * channels is non backlog, the combined RecordAttributes is non backlog.
+ *
+ * <p>Currently, we only support switching the backlog status from null to backlog and backlog to
+ * non-backlog. Switching from non-backlog to backlog is not support at the moment, and it will be
+ * ignored.
+ */
+public class RecordAttributesValve {
Review Comment:
1. The name doesn't really reflect what the class does. I'd suggest something like `RecordAttributesCombiner`.
2. Note that `RecordAttributes` is a general-purpose class for carrying not only `isBacklog` but more possible attributes in future. Therefore, the class-wise JavaDoc should be more focused on how in general `RecordAttributes` are combined. The detailed strategy for combining individual attributes can be documented elsewhere, e.g. a dedicated private method.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/StreamTaskNetworkInputTest.java:
##########
@@ -318,6 +320,56 @@ public void testBatchProcessingRecordsCanBeInterrupted() throws Exception {
assertThat(output.getNumberOfEmittedRecords()).isEqualTo(expectedElementCount * 2);
}
+ @Test
+ public void testProcessRecordAttributes() throws Exception {
+ int numInputChannels = 2;
+ LongSerializer inSerializer = LongSerializer.INSTANCE;
+ StreamTestSingleInputGate<Long> inputGate =
+ new StreamTestSingleInputGate<>(numInputChannels, 0, inSerializer, 1024);
+ StreamTaskNetworkInput<Long> input =
+ new StreamTaskNetworkInput<>(
+ createCheckpointedInputGate(inputGate.getInputGate()),
+ inSerializer,
+ ioManager,
+ new StatusWatermarkValve(numInputChannels),
+ 0,
+ () -> true);
+
+ final CollectingDataOutput<Long> output = new CollectingDataOutput<>();
+
+ inputGate.sendElement(
+ new RecordAttributesBuilder(Collections.emptyList()).setBacklog(true).build(), 0);
+ input.emitNext(output);
+ assertThat(output.getEvents()).isEmpty();
+
+ inputGate.sendElement(new StreamRecord<>(0L, 0), 1);
+ inputGate.sendElement(
+ new RecordAttributesBuilder(Collections.emptyList()).setBacklog(true).build(), 1);
+ inputGate.sendElement(new StreamRecord<>(1L, 1), 1);
+ inputGate.sendElement(new StreamRecord<>(2L, 2), 1);
+ input.emitNext(output);
+ assertThat(output.getEvents())
+ .containsExactly(
+ new StreamRecord<>(0L, 0),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(true)
+ .build());
Review Comment:
Minor:
I'd suggest to make these StreamRecord / RecordAttributes local variables, rather that creating duplicated objects for `sendElement` and assertions.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/TimestampsAndWatermarks.java:
##########
@@ -80,6 +80,9 @@ ReaderOutput<T> createMainOutput(
/** Stops emitting periodic watermarks. */
void stopPeriodicWatermarkEmits();
+ /** Trigger a periodicEmit immediately. */
+ default void triggerPeriodicEmit(long wallClockTimestamp) {}
Review Comment:
Why providing a default implementation?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/SortingBacklogDataInput.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators.sort;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.AlgorithmOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.io.AvailabilityProvider;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.jobgraph.tasks.TaskInvokable;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.sort.ExternalSorter;
+import org.apache.flink.runtime.operators.sort.PushSorter;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.DataInputStatus;
+import org.apache.flink.streaming.runtime.io.StreamTaskInput;
+import org.apache.flink.streaming.runtime.streamrecord.InternalRecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * SortingBacklogDataInput forwards the input records to output util it switch to backlog
+ * processing. During backlog processing, it buffers and sorts the input records and outputs the
+ * sorted records to the downstream when switching to non backlog processing.
+ */
+@Internal
+public class SortingBacklogDataInput<T, K> implements StreamTaskInput<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SortingBacklogDataInput.class);
+
+ private final StreamTaskInput<T> wrappedInput;
+ private final PushSorter<Tuple2<byte[], StreamRecord<T>>> sorter;
+ private final KeySelector<T, K> keySelector;
+ private final TypeSerializer<K> keySerializer;
+ private final DataOutputSerializer dataOutputSerializer;
Review Comment:
Minor: These fields can be inner to `SortingDataOutput`.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/RecordAttributesValve.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput.DataOutput;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+
+/**
+ * RecordAttributesValve combine RecordAttributes from different input channels. If any of the input
+ * channels is non backlog, the combined RecordAttributes is non backlog.
+ *
+ * <p>Currently, we only support switching the backlog status from null to backlog and backlog to
+ * non-backlog. Switching from non-backlog to backlog is not support at the moment, and it will be
+ * ignored.
+ */
+public class RecordAttributesValve {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RecordAttributesValve.class);
+
+ private final int numInputChannels;
+ private final RecordAttributes[] allChannelRecordAttributes;
+
+ private int backlogChannelsCnt = 0;
+ private RecordAttributes lastOutputAttributes = null;
+
+ public RecordAttributesValve(int numInputChannels) {
+ this.numInputChannels = numInputChannels;
+ this.allChannelRecordAttributes = new RecordAttributes[numInputChannels];
+ }
+
+ public void inputRecordAttributes(
+ RecordAttributes recordAttributes, int channelIdx, DataOutput<?> output)
+ throws Exception {
+ LOG.debug("RecordAttributes: {} from channel idx: {}", recordAttributes, channelIdx);
+ RecordAttributes lastChannelRecordAttributes = allChannelRecordAttributes[channelIdx];
+ allChannelRecordAttributes[channelIdx] = recordAttributes;
+
+ if (lastChannelRecordAttributes == null) {
+ lastChannelRecordAttributes =
+ new RecordAttributesBuilder(Collections.emptyList()).build();
+ }
+
+ if (lastChannelRecordAttributes.isBacklog() == recordAttributes.isBacklog()) {
+ return;
+ }
+
+ if (recordAttributes.isBacklog()) {
+ backlogChannelsCnt += 1;
+ } else {
+ backlogChannelsCnt -= 1;
+ }
+
+ if (lastOutputAttributes == null && backlogChannelsCnt != numInputChannels) {
+ return;
+ }
+
+ final RecordAttributesBuilder builder =
+ new RecordAttributesBuilder(Collections.emptyList());
+ builder.setBacklog(backlogChannelsCnt >= numInputChannels);
Review Comment:
This means iff all input channels are in `isBacklog=true` status will the task see `isBacklog=true`. Why is that? Did I missed anything from the FLIP?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/BacklogUtils.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.translators;
+
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.graph.StreamNode;
+import org.apache.flink.streaming.api.graph.TransformationTranslator;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL_DURING_BACKLOG;
+
+/** A utility class for applying sorting inputs during backlog processing. */
+public class BacklogUtils {
+ public static boolean isCheckpointDisableDuringBacklog(
+ TransformationTranslator.Context context) {
+ return context.getGraphGeneratorConfig().get(CHECKPOINTING_INTERVAL_DURING_BACKLOG) != null
+ && context.getGraphGeneratorConfig()
+ .get(CHECKPOINTING_INTERVAL_DURING_BACKLOG)
+ .isZero();
+ }
+
+ public static void applyBacklogProcessingSettings(
+ Transformation<?> transformation,
+ TransformationTranslator.Context context,
+ StreamNode node) {
+ node.addInputRequirement(0, StreamConfig.InputRequirement.SORTED_DURING_BACKLOG);
+ Map<ManagedMemoryUseCase, Integer> operatorScopeUseCaseWeights = new HashMap<>();
+ Integer operatorMemoryWeights =
+ transformation
+ .getManagedMemoryOperatorScopeUseCaseWeights()
+ .get(ManagedMemoryUseCase.OPERATOR);
+ operatorScopeUseCaseWeights.put(
+ ManagedMemoryUseCase.OPERATOR,
+ operatorMemoryWeights == null
+ ? deriveMemoryWeight(context.getGraphGeneratorConfig())
+ : operatorMemoryWeights);
Review Comment:
Does this mean if the operator is already configured with some managed memory, we will not require more memory for the sorter? Then what if both the operator and the sorting input needs managed memory? How do they share the memory?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java:
##########
@@ -415,6 +417,12 @@ private void configureStreamGraphStreaming(final StreamGraph graph) {
graph.setCheckpointStorage(checkpointStorage);
graph.setSavepointDirectory(savepointDir);
graph.setGlobalStreamExchangeMode(deriveGlobalStreamExchangeModeStreaming());
+
+ if (Duration.ZERO.equals(
+ configuration.get(
+ ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL_DURING_BACKLOG))) {
Review Comment:
Same here for deciding whether to enable the mixed mode.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalBacklogAwareTimerServiceImpl.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+/**
+ * InternalBacklogAwareTimerServiceImpl is the implementation of InternalTimerService that manages
+ * event time timers differently under different backlog status. It manages the processing time
+ * timers in the same way as the {@link InternalTimerServiceImpl}.
+ *
+ * <p>The backlog status is false at the beginning. When backlog status is false, the timer services
+ * is the same as the {@link InternalTimerServiceImpl}, where event time timers are triggered as the
+ * watermark advancing.
+ *
+ * <p>When backlog status switch to true, the timer service manages event time timer with a single
+ * active key at a time. When the active key changed, all the event time timers of the last key are
+ * triggered up to the maximum watermark during backlog processing.
+ *
+ * <p>When the backlog status switch from true to false, all the event time timers of the last key
+ * are triggered up to the maximum watermark during backlog processing. And the current watermark of
+ * the timer service will be the maximum watermark during backlog processing.
+ */
+@Internal
+public class InternalBacklogAwareTimerServiceImpl<K, N> implements InternalTimerService<N> {
Review Comment:
TBH, I'm not a big fan of inherit. The problem is that, by directly accessing protected fields of the superclass from the subclasses, the contract between the classes becomes obscure. As subclass can see internals of the superclass, changes to the superclass can easily break things that subclasses depend on thus cause problems, and such issues are very hard to avoid.
In many cases, an inherit approach can be replaced by an equivalent approach based on composition, which has clearer contracts between classes. But admittedly it's no always possible and sometimes gets costly.
For the classes below, I guess I haven't spent enough time to explore whether it's possible to switch to a composition based approach. Do you think it's feasible?
- `BacklogTimeService`
- `BatchExecutionInternalTimeService`
- `InternalBacklogAwareTimerServiceManagerImpl`
- `InternalTimeServiceManagerImpl`
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/RecordAttributesBuilder.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.streamrecord;
+
+import org.apache.flink.annotation.Experimental;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/** The builder class for {@link RecordAttributes}. */
+@Experimental
+public class RecordAttributesBuilder {
+ private static final Logger LOG = LoggerFactory.getLogger(RecordAttributesBuilder.class);
+ private final List<RecordAttributes> lastRecordAttributesOfInputs;
+ @Nullable private Boolean backlog = null;
+
+ /**
+ * This constructor takes a list of the last recordAttributes received from each of the
+ * operator's inputs. When this list is not empty, it will be used to determine the default
Review Comment:
By "each of the operator's input", does this mean logical or physical inputs? Asked differently, is the list size corresponds to number of input edges of the job graph or execution graph?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java:
##########
@@ -396,6 +396,10 @@ public int numEventTimeTimers(N namespace) {
return countTimersInNamespaceInternal(namespace, eventTimeTimersQueue);
}
+ public KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> getEventTimeTimersQueue() {
+ return eventTimeTimersQueue;
+ }
Review Comment:
It feels hacky to expose an internal queue from one component and use it to construct another component. If the queue is meant to be shared, I think the correct way is to create the queue out side the two components and pass it in as constructor arguments.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/BacklogTimeService.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.streaming.api.operators.sorted.state.BatchExecutionInternalTimeService;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * An implementation of a {@link InternalTimerService} that manages timers with a single active key
+ * at a time. This is used by {@link
+ * org.apache.flink.streaming.api.operators.InternalBacklogAwareTimerServiceImpl} during backlog
+ * processing.
+ */
+class BacklogTimeService<K, N> extends BatchExecutionInternalTimeService<K, N> {
+ private long maxWatermarkDuringBacklog;
+
+ public BacklogTimeService(
+ ProcessingTimeService processingTimeService,
+ Triggerable<K, N> triggerTarget,
+ KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> eventTimeTimersQueue) {
+ super(processingTimeService, triggerTarget, eventTimeTimersQueue, null);
+ }
+
+ @Override
+ public void registerProcessingTimeTimer(N namespace, long time) {
+ throw new UnsupportedOperationException(
+ "BacklogTimeService does not support registering processing timer.");
+ }
+
+ @Override
+ public void deleteProcessingTimeTimer(N namespace, long time) {
+ throw new UnsupportedOperationException(
+ "BacklogTimeService does not support deleting processing timer.");
+ }
+
+ /**
+ * Set the current key of the time service. If the new key is different from the last key, all
+ * the event time timers of the last key whose timestamp is less than or equal to the max
+ * watermark during backlog are triggered.
+ */
+ public void setCurrentKey(K newKey) throws Exception {
+ if (newKey != null && newKey.equals(currentKey)) {
+ return;
+ }
+
+ TimerHeapInternalTimer<K, N> timer;
+ List<TimerHeapInternalTimer<K, N>> skippedTimers = new LinkedList<>();
+ if (currentKey != null) {
+ while ((timer = eventTimeTimersQueue.peek()) != null
+ && timer.getTimestamp() <= maxWatermarkDuringBacklog) {
+ eventTimeTimersQueue.poll();
+
+ if (timer.getKey() != currentKey) {
+ skippedTimers.add(timer);
+ } else {
+ triggerTarget.onEventTime(timer);
+ }
+ }
+ eventTimeTimersQueue.addAll(skippedTimers);
+ }
+
+ if (newKey == null) {
+ currentWatermark = maxWatermarkDuringBacklog;
+ }
Review Comment:
What does `newKey == null` mean? And why we need to update the current watermark under such condition?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/RecordAttributesBuilder.java:
##########
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.streamrecord;
+
+import org.apache.flink.annotation.Experimental;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/** The builder class for {@link RecordAttributes}. */
+@Experimental
+public class RecordAttributesBuilder {
+ private static final Logger LOG = LoggerFactory.getLogger(RecordAttributesBuilder.class);
+ private final List<RecordAttributes> lastRecordAttributesOfInputs;
+ @Nullable private Boolean backlog = null;
+
+ /**
+ * This constructor takes a list of the last recordAttributes received from each of the
+ * operator's inputs. When this list is not empty, it will be used to determine the default
Review Comment:
Either way, it would be better to clearly explain this in JavaDoc.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/RecordAttributesValve.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput.DataOutput;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+
+/**
+ * RecordAttributesValve combine RecordAttributes from different input channels. If any of the input
+ * channels is non backlog, the combined RecordAttributes is non backlog.
+ *
+ * <p>Currently, we only support switching the backlog status from null to backlog and backlog to
+ * non-backlog. Switching from non-backlog to backlog is not support at the moment, and it will be
+ * ignored.
+ */
+public class RecordAttributesValve {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RecordAttributesValve.class);
+
+ private final int numInputChannels;
+ private final RecordAttributes[] allChannelRecordAttributes;
+
+ private int backlogChannelsCnt = 0;
+ private RecordAttributes lastOutputAttributes = null;
+
+ public RecordAttributesValve(int numInputChannels) {
+ this.numInputChannels = numInputChannels;
+ this.allChannelRecordAttributes = new RecordAttributes[numInputChannels];
+ }
+
+ public void inputRecordAttributes(
+ RecordAttributes recordAttributes, int channelIdx, DataOutput<?> output)
+ throws Exception {
+ LOG.debug("RecordAttributes: {} from channel idx: {}", recordAttributes, channelIdx);
+ RecordAttributes lastChannelRecordAttributes = allChannelRecordAttributes[channelIdx];
+ allChannelRecordAttributes[channelIdx] = recordAttributes;
+
+ if (lastChannelRecordAttributes == null) {
+ lastChannelRecordAttributes =
+ new RecordAttributesBuilder(Collections.emptyList()).build();
+ }
+
+ if (lastChannelRecordAttributes.isBacklog() == recordAttributes.isBacklog()) {
+ return;
+ }
+
+ if (recordAttributes.isBacklog()) {
+ backlogChannelsCnt += 1;
+ } else {
+ backlogChannelsCnt -= 1;
+ }
+
+ if (lastOutputAttributes == null && backlogChannelsCnt != numInputChannels) {
+ return;
+ }
+
+ final RecordAttributesBuilder builder =
+ new RecordAttributesBuilder(Collections.emptyList());
+ builder.setBacklog(backlogChannelsCnt >= numInputChannels);
+ final RecordAttributes outputAttribute = builder.build();
+ if (lastOutputAttributes == null
+ || lastOutputAttributes.isBacklog() != outputAttribute.isBacklog()) {
+ if (lastOutputAttributes != null && !lastOutputAttributes.isBacklog()) {
+ LOG.warn(
+ "Switching from non-backlog to backlog is currently not supported. Backlog status remains.");
+ return;
+ }
Review Comment:
The warning can easily get ignored. I'd suggest to fail fast.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/source/ProgressiveTimestampsAndWatermarks.java:
##########
@@ -152,7 +152,7 @@ public void stopPeriodicWatermarkEmits() {
}
}
- void triggerPeriodicEmit(@SuppressWarnings("unused") long wallClockTimestamp) {
+ public void triggerPeriodicEmit(@SuppressWarnings("unused") long wallClockTimestamp) {
Review Comment:
`@Override`
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/SortingBacklogDataInput.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators.sort;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.AlgorithmOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.io.AvailabilityProvider;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.jobgraph.tasks.TaskInvokable;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.sort.ExternalSorter;
+import org.apache.flink.runtime.operators.sort.PushSorter;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.DataInputStatus;
+import org.apache.flink.streaming.runtime.io.StreamTaskInput;
+import org.apache.flink.streaming.runtime.streamrecord.InternalRecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * SortingBacklogDataInput forwards the input records to output util it switch to backlog
+ * processing. During backlog processing, it buffers and sorts the input records and outputs the
+ * sorted records to the downstream when switching to non backlog processing.
+ */
+@Internal
+public class SortingBacklogDataInput<T, K> implements StreamTaskInput<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SortingBacklogDataInput.class);
+
+ private final StreamTaskInput<T> wrappedInput;
+ private final PushSorter<Tuple2<byte[], StreamRecord<T>>> sorter;
+ private final KeySelector<T, K> keySelector;
+ private final TypeSerializer<K> keySerializer;
+ private final DataOutputSerializer dataOutputSerializer;
+ private final SortingDataOutput sortingDataOutput;
+ private final StreamTask.CanEmitBatchOfRecordsChecker canEmitBatchOfRecords;
+ private MutableObjectIterator<Tuple2<byte[], StreamRecord<T>>> sortedInput = null;
+ private long watermarkSeenDuringBacklog = Long.MIN_VALUE;
+ private WatermarkStatus lastWatermarkStatusDuringBacklog = null;
+ private volatile OperatingMode mode = OperatingMode.PROCESSING_REALTIME;
+
+ private enum OperatingMode {
+
+ // We are processing realtime data, all the input data pass through immediately.
Review Comment:
Minor: Use JavaDoc, so that IDE prompts the documents when the cursor stays over a enum value.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/InternalRecordAttributes.java:
##########
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.streamrecord;
+
+import org.apache.flink.annotation.Internal;
+
+import java.util.Objects;
+
+/** The RecordAttributes that contains extra information to be used internally. */
+@Internal
+public class InternalRecordAttributes extends RecordAttributes {
+ private final long maxWatermarkDuringBacklog;
+
+ public InternalRecordAttributes(boolean backlog, long maxWatermarkDuringBacklog) {
+ super(backlog);
+ this.maxWatermarkDuringBacklog = maxWatermarkDuringBacklog;
+ }
+
+ public long getMaxWatermarkDuringBacklog() {
+ return maxWatermarkDuringBacklog;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ InternalRecordAttributes that = (InternalRecordAttributes) o;
+ return maxWatermarkDuringBacklog == that.maxWatermarkDuringBacklog;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), maxWatermarkDuringBacklog);
+ }
+
+ @Override
+ public String toString() {
+ return "InternalRecordAttributes{"
+ + "backlog="
+ + isBacklog()
Review Comment:
This is fragile because when updating the superclass developers may not notice they need to also change the subclass.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/SortingBacklogDataInput.java:
##########
@@ -0,0 +1,330 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators.sort;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.AlgorithmOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.io.AvailabilityProvider;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.jobgraph.tasks.TaskInvokable;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.sort.ExternalSorter;
+import org.apache.flink.runtime.operators.sort.PushSorter;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.DataInputStatus;
+import org.apache.flink.streaming.runtime.io.StreamTaskInput;
+import org.apache.flink.streaming.runtime.streamrecord.InternalRecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * SortingBacklogDataInput forwards the input records to output util it switch to backlog
+ * processing. During backlog processing, it buffers and sorts the input records and outputs the
+ * sorted records to the downstream when switching to non backlog processing.
+ */
+@Internal
+public class SortingBacklogDataInput<T, K> implements StreamTaskInput<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SortingBacklogDataInput.class);
+
+ private final StreamTaskInput<T> wrappedInput;
+ private final PushSorter<Tuple2<byte[], StreamRecord<T>>> sorter;
+ private final KeySelector<T, K> keySelector;
+ private final TypeSerializer<K> keySerializer;
+ private final DataOutputSerializer dataOutputSerializer;
+ private final SortingDataOutput sortingDataOutput;
+ private final StreamTask.CanEmitBatchOfRecordsChecker canEmitBatchOfRecords;
+ private MutableObjectIterator<Tuple2<byte[], StreamRecord<T>>> sortedInput = null;
+ private long watermarkSeenDuringBacklog = Long.MIN_VALUE;
+ private WatermarkStatus lastWatermarkStatusDuringBacklog = null;
+ private volatile OperatingMode mode = OperatingMode.PROCESSING_REALTIME;
+
+ private enum OperatingMode {
+
+ // We are processing realtime data, all the input data pass through immediately.
+ PROCESSING_REALTIME,
+
+ // We are processing backlog data, all the input data are put in the sorter.
+ SORTING_BACKLOG,
+
+ // We are switching from backlog processing to realtime processing, the data input the
+ // sorter are flushing to the downstream.
+ FLUSHING_BACKLOG
Review Comment:
Trying to understand this. The reason we have a mode for switching from backlog to realtime but not for the other way around is that, switching from realtime to backlog is expected to be completed immediately, while switching from backlog to realtime takes time for emitting the buffered / sorted results. Is that correct?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/RecordAttributesValve.java:
##########
@@ -0,0 +1,93 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput.DataOutput;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+
+/**
+ * RecordAttributesValve combine RecordAttributes from different input channels. If any of the input
+ * channels is non backlog, the combined RecordAttributes is non backlog.
+ *
+ * <p>Currently, we only support switching the backlog status from null to backlog and backlog to
+ * non-backlog. Switching from non-backlog to backlog is not support at the moment, and it will be
+ * ignored.
+ */
+public class RecordAttributesValve {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RecordAttributesValve.class);
+
+ private final int numInputChannels;
+ private final RecordAttributes[] allChannelRecordAttributes;
+
+ private int backlogChannelsCnt = 0;
+ private RecordAttributes lastOutputAttributes = null;
+
+ public RecordAttributesValve(int numInputChannels) {
+ this.numInputChannels = numInputChannels;
+ this.allChannelRecordAttributes = new RecordAttributes[numInputChannels];
+ }
+
+ public void inputRecordAttributes(
+ RecordAttributes recordAttributes, int channelIdx, DataOutput<?> output)
+ throws Exception {
+ LOG.debug("RecordAttributes: {} from channel idx: {}", recordAttributes, channelIdx);
+ RecordAttributes lastChannelRecordAttributes = allChannelRecordAttributes[channelIdx];
+ allChannelRecordAttributes[channelIdx] = recordAttributes;
+
+ if (lastChannelRecordAttributes == null) {
+ lastChannelRecordAttributes =
+ new RecordAttributesBuilder(Collections.emptyList()).build();
+ }
Review Comment:
Minor:
1. Why not initialize the array with empty record-attributes, so that we avoid this if-statement for each incoming record-attributes?
2. It seems `new RecordAttributesBuilder(Collections.emptyList()).build()` is used a lot. We may consider introduce a static singleton instance, to avoid unnecessary object creations.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/OneInputTransformationTranslator.java:
##########
@@ -62,12 +63,19 @@ public Collection<Integer> translateForBatchInternal(
@Override
public Collection<Integer> translateForStreamingInternal(
final OneInputTransformation<IN, OUT> transformation, final Context context) {
- return translateInternal(
- transformation,
- transformation.getOperatorFactory(),
- transformation.getInputType(),
- transformation.getStateKeySelector(),
- transformation.getStateKeyType(),
- context);
+ final KeySelector<IN, ?> keySelector = transformation.getStateKeySelector();
+ final Collection<Integer> ids =
+ translateInternal(
+ transformation,
+ transformation.getOperatorFactory(),
+ transformation.getInputType(),
+ keySelector,
+ transformation.getStateKeyType(),
+ context);
+ final StreamNode node = context.getStreamGraph().getStreamNode(transformation.getId());
+ if (keySelector != null && BacklogUtils.isCheckpointDisableDuringBacklog(context)) {
+ BacklogUtils.applyBacklogProcessingSettings(transformation, context, node);
+ }
Review Comment:
I'd suggest to decouple whether to enable mixed mode from the checkpoint-during-backlog as much as possible. As I suggested in the FLIP discussion, it is only a current strategy to enable the mixed mode when checkpointing.interval-during-backlog == 0 (&& runtime-mode == streaming). However, this may change in future. It would be nice to have a centralized position to decide whether to enable the mixed mode (from whatever inputs it needs), and just read the result from everywhere else.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/AbstractStreamTaskNetworkInput.java:
##########
@@ -141,19 +145,34 @@ public DataInputStatus emitNext(DataOutput<T> output) throws Exception {
}
}
- private void processElement(StreamElement recordOrMark, DataOutput<T> output) throws Exception {
- if (recordOrMark.isRecord()) {
- output.emitRecord(recordOrMark.asRecord());
- } else if (recordOrMark.isWatermark()) {
+ /**
+ * Process the given stream element and returns whether to continue processing the next stream
+ * element without returning from the emitNext method.
+ */
+ private boolean processElement(StreamElement streamElement, DataOutput<T> output)
+ throws Exception {
+ if (streamElement.isRecord()) {
+ output.emitRecord(streamElement.asRecord());
+ return true;
+ } else if (streamElement.isWatermark()) {
statusWatermarkValve.inputWatermark(
- recordOrMark.asWatermark(), flattenedChannelIndices.get(lastChannel), output);
- } else if (recordOrMark.isLatencyMarker()) {
- output.emitLatencyMarker(recordOrMark.asLatencyMarker());
- } else if (recordOrMark.isWatermarkStatus()) {
+ streamElement.asWatermark(), flattenedChannelIndices.get(lastChannel), output);
+ return true;
+ } else if (streamElement.isLatencyMarker()) {
+ output.emitLatencyMarker(streamElement.asLatencyMarker());
+ return true;
+ } else if (streamElement.isWatermarkStatus()) {
statusWatermarkValve.inputWatermarkStatus(
- recordOrMark.asWatermarkStatus(),
+ streamElement.asWatermarkStatus(),
flattenedChannelIndices.get(lastChannel),
output);
+ return true;
+ } else if (streamElement.isRecordAttributes()) {
+ recordAttributesValve.inputRecordAttributes(
+ streamElement.asRecordAttributes(),
+ flattenedChannelIndices.get(lastChannel),
+ output);
+ return false;
Review Comment:
IIUC, the purpose here is to break the record processing and provide operators a chance to change their behavior right after the record-attribute (i.e. change of the isBacklog status).
To make this easier to understand, I'd suggest:
- Update the JavaDoc for this method, explain why we would want to stop processing and return from the emitNext method.
- rename the local variable `canContinue` in `emitNext` to something like `breakBatchRecordEmitting`.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/BacklogTimeService.java:
##########
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.streaming.api.operators.sorted.state.BatchExecutionInternalTimeService;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * An implementation of a {@link InternalTimerService} that manages timers with a single active key
+ * at a time. This is used by {@link
+ * org.apache.flink.streaming.api.operators.InternalBacklogAwareTimerServiceImpl} during backlog
+ * processing.
+ */
+class BacklogTimeService<K, N> extends BatchExecutionInternalTimeService<K, N> {
+ private long maxWatermarkDuringBacklog;
+
+ public BacklogTimeService(
+ ProcessingTimeService processingTimeService,
+ Triggerable<K, N> triggerTarget,
+ KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> eventTimeTimersQueue) {
+ super(processingTimeService, triggerTarget, eventTimeTimersQueue, null);
+ }
+
+ @Override
+ public void registerProcessingTimeTimer(N namespace, long time) {
+ throw new UnsupportedOperationException(
+ "BacklogTimeService does not support registering processing timer.");
+ }
+
+ @Override
+ public void deleteProcessingTimeTimer(N namespace, long time) {
+ throw new UnsupportedOperationException(
+ "BacklogTimeService does not support deleting processing timer.");
+ }
+
+ /**
+ * Set the current key of the time service. If the new key is different from the last key, all
+ * the event time timers of the last key whose timestamp is less than or equal to the max
+ * watermark during backlog are triggered.
+ */
+ public void setCurrentKey(K newKey) throws Exception {
Review Comment:
`@Override`
##########
flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/TimeTestUtil.scala:
##########
@@ -98,6 +98,8 @@ object TimeTestUtil {
}
}
+ override def processRecordAttributes(recordAttributes: RecordAttributes): Unit =
+ super.processRecordAttributes(recordAttributes)
Review Comment:
I'm not familiar with Scala. What is this for? Shouldn't the method from the super class be called anyway if we don't override it?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/OneInputTransformationTranslator.java:
##########
@@ -62,12 +63,19 @@ public Collection<Integer> translateForBatchInternal(
@Override
public Collection<Integer> translateForStreamingInternal(
final OneInputTransformation<IN, OUT> transformation, final Context context) {
- return translateInternal(
- transformation,
- transformation.getOperatorFactory(),
- transformation.getInputType(),
- transformation.getStateKeySelector(),
- transformation.getStateKeyType(),
- context);
+ final KeySelector<IN, ?> keySelector = transformation.getStateKeySelector();
+ final Collection<Integer> ids =
+ translateInternal(
+ transformation,
+ transformation.getOperatorFactory(),
+ transformation.getInputType(),
+ keySelector,
+ transformation.getStateKeyType(),
+ context);
+ final StreamNode node = context.getStreamGraph().getStreamNode(transformation.getId());
+ if (keySelector != null && BacklogUtils.isCheckpointDisableDuringBacklog(context)) {
+ BacklogUtils.applyBacklogProcessingSettings(transformation, context, node);
+ }
Review Comment:
Same for `ReduceTransformationTranslator`.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-2597167245
This PR is being marked as stale since it has not had any activity in the last 180 days.
If you would like to keep this PR alive, please leave a comment asking for a review.
If the PR has merge conflicts, update it with the latest from the base branch.
If you are having difficulty finding a reviewer, please reach out to the
community, contact details can be found here: https://flink.apache.org/what-is-flink/community/
If this PR is no longer valid or desired, please feel free to close it.
If no activity occurs in the next 90 days, it will be automatically closed.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33202][runtime] Support switching from batch to stream mode to improve throughput when processing backlog data [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-1761060678
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "adb943cdb234ee3c61bc841706598be7cad1ae92",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "adb943cdb234ee3c61bc841706598be7cad1ae92",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* adb943cdb234ee3c61bc841706598be7cad1ae92 UNKNOWN
<details>
<summary>Bot commands</summary>
The @flinkbot bot supports the following commands:
- `@flinkbot run azure` re-run the last Azure build
</details>
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1384473467
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/StreamElementSerializer.java:
##########
@@ -146,6 +147,8 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
target.writeLong(source.readLong());
target.writeLong(source.readLong());
target.writeInt(source.readInt());
+ } else if (tag == TAG_RECORD_ATTRIBUTES) {
+ target.writeBoolean(source.readBoolean());
Review Comment:
`InternalRecordAttributes` should only be used internally in an operator chain. It is not expected to be serialized and deserialized.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33202][runtime] Support switching from batch to stream mode to improve throughput when processing backlog data [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1366613177
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/MultiInputTransformationTranslator.java:
##########
@@ -64,7 +64,8 @@ protected Collection<Integer> translateForBatchInternal(
IntStream.range(0, inputs.size())
.mapToObj(
idx -> {
- if (keySelectors.get(idx) != null) {
+ if (keySelectors.get(idx) != null
+ && !transformation.isInternalSorterSupported()) {
return StreamConfig.InputRequirement.SORTED;
} else {
Review Comment:
`SORTING_DURING_BACKLOG` is only supported in StreamOperator with one input according to the FLIP.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33202][runtime] Support switching from batch to stream mode to improve throughput when processing backlog data [flink]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1371166494
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/StreamElementSerializer.java:
##########
@@ -146,6 +147,8 @@ public void copy(DataInputView source, DataOutputView target) throws IOException
target.writeLong(source.readLong());
target.writeLong(source.readLong());
target.writeInt(source.readInt());
+ } else if (tag == TAG_RECORD_ATTRIBUTES) {
+ target.writeBoolean(source.readBoolean());
Review Comment:
Do we need to serialize/deserialize InternalRecordAttributes as well?
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/InternalBacklogAwareTimerServiceImplTest.java:
##########
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueSetFactory;
+import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.streaming.api.operators.TimeServiceTestUtils.createBacklogTimerService;
+import static org.apache.flink.streaming.api.operators.TimeServiceTestUtils.createInternalTimerService;
+import static org.apache.flink.streaming.api.operators.TimeServiceTestUtils.createTimerQueue;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link InternalBacklogAwareTimerServiceImpl}. */
+public class InternalBacklogAwareTimerServiceImplTest extends InternalTimerServiceImplTest {
Review Comment:
Let's override the way `InternalTimerServiceImplTest` uses to create timer service into producing `InternalBacklogAwareTimerServiceImpl` as well. Otherwise extending `InternalTimerServiceImplTest` will not test `InternalBacklogAwareTimerServiceImpl` against more test cases.
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/StreamingWithBacklogITCase.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.streaming.api.datastream;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.AggregateFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.hybrid.HybridSource;
+import org.apache.flink.connector.datagen.source.DataGeneratorSource;
+import org.apache.flink.connector.datagen.source.GeneratorFunction;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL_DURING_BACKLOG;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Integration test for streaming job with backlog. */
+public class StreamingWithBacklogITCase {
+ @Test
+ void testKeyedAggregationWithBacklog() throws Exception {
+ final Configuration config = new Configuration();
+ config.set(CHECKPOINTING_INTERVAL_DURING_BACKLOG, Duration.ZERO);
Review Comment:
The test cases in this class can also pass without this configuration, which means we are not sure whether the changes in this PR took effect. Would it be better if we add some verifications that are exclusive to backlog cases? For example, the frequency of switching keys is much lower than normal cases, and the watermark happens to stop advancing until backlog status changes to false.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/LambdaTrigger.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.assertj.core.api.Assertions;
+
+import java.util.function.Consumer;
+
+/** Triggerable for test. */
+public class LambdaTrigger<K, N> implements Triggerable<K, N> {
Review Comment:
It might be better to improve the naming and JavaDoc of this class, since it is a shared utility class instead of a private class now.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33202][runtime] Support switching from batch to stream mode to improve throughput when processing backlog data [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1369928452
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/SortingBacklogDataInputTest.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators.sort;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.operators.testutils.DummyInvokable;
+import org.apache.flink.runtime.operators.testutils.MockEnvironment;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.DataInputStatus;
+import org.apache.flink.streaming.runtime.streamrecord.InternalRecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+/** Tests for {@link SortingBacklogDataInput}. */
+public class SortingBacklogDataInputTest {
+ @Test
+ public void sortingDuringBacklog() throws Exception {
+ CollectingDataOutput<Integer> collectingDataOutput = new CollectingDataOutput<>();
+ CollectionDataInput<Integer> input =
+ new CollectionDataInput<>(
+ Arrays.asList(
+ new StreamRecord<>(2, 0),
+ new StreamRecord<>(1, 0),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(true)
+ .build(),
+ new StreamRecord<>(1, 3),
+ new StreamRecord<>(1, 1),
+ new StreamRecord<>(2, 1),
+ new StreamRecord<>(2, 3),
+ new StreamRecord<>(1, 2),
+ new StreamRecord<>(2, 2),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(false)
+ .build(),
+ new StreamRecord<>(1, 4),
+ new StreamRecord<>(2, 4),
+ new StreamRecord<>(1, 5)));
+ MockEnvironment environment = MockEnvironment.builder().build();
+ SortingBacklogDataInput<Integer, Integer> sortingDataInput =
+ new SortingBacklogDataInput<>(
+ input,
+ new IntSerializer(),
+ new IntSerializer(),
+ (KeySelector<Integer, Integer>) value -> value,
+ environment.getMemoryManager(),
+ environment.getIOManager(),
+ true,
+ 1.0,
+ new Configuration(),
+ new DummyInvokable(),
+ new ExecutionConfig(),
+ () -> true);
+
+ DataInputStatus inputStatus;
+ do {
+ inputStatus = sortingDataInput.emitNext(collectingDataOutput);
+ } while (inputStatus != DataInputStatus.END_OF_INPUT);
+
+ org.assertj.core.api.Assertions.assertThat(collectingDataOutput.events)
+ .containsExactly(
+ new StreamRecord<>(2, 0),
+ new StreamRecord<>(1, 0),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(true)
+ .build(),
+ new InternalRecordAttributes(true, Long.MIN_VALUE),
+ new StreamRecord<>(1, 1),
+ new StreamRecord<>(1, 2),
+ new StreamRecord<>(1, 3),
+ new StreamRecord<>(2, 1),
+ new StreamRecord<>(2, 2),
+ new StreamRecord<>(2, 3),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(false)
+ .build(),
+ new StreamRecord<>(1, 4),
+ new StreamRecord<>(2, 4),
+ new StreamRecord<>(1, 5));
+ }
+
+ @Test
+ public void watermarkPropagation() throws Exception {
+ CollectingDataOutput<Integer> collectingDataOutput = new CollectingDataOutput<>();
+ CollectionDataInput<Integer> input =
+ new CollectionDataInput<>(
+ Arrays.asList(
+ new StreamRecord<>(1, 3),
+ new Watermark(1),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(true)
+ .build(),
+ new StreamRecord<>(1, 1),
+ new Watermark(2),
+ new StreamRecord<>(2, 1),
+ new Watermark(3),
+ new StreamRecord<>(2, 3),
+ new Watermark(4),
+ new StreamRecord<>(1, 2),
+ new Watermark(5),
+ new StreamRecord<>(2, 2),
+ new Watermark(6),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(false)
+ .build()));
+ MockEnvironment environment = MockEnvironment.builder().build();
+ SortingBacklogDataInput<Integer, Integer> sortingDataInput =
+ new SortingBacklogDataInput<>(
+ input,
+ new IntSerializer(),
+ new IntSerializer(),
+ (KeySelector<Integer, Integer>) value -> value,
+ environment.getMemoryManager(),
+ environment.getIOManager(),
+ true,
+ 1.0,
+ new Configuration(),
+ new DummyInvokable(),
+ new ExecutionConfig(),
+ () -> true);
+
+ DataInputStatus inputStatus;
+ do {
+ inputStatus = sortingDataInput.emitNext(collectingDataOutput);
+ } while (inputStatus != DataInputStatus.END_OF_INPUT);
+
+ org.assertj.core.api.Assertions.assertThat(collectingDataOutput.events)
+ .containsExactly(
+ new StreamRecord<>(1, 3),
+ new Watermark(1),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(true)
+ .build(),
+ new InternalRecordAttributes(true, 6L),
+ new StreamRecord<>(1, 1),
+ new StreamRecord<>(1, 2),
+ new StreamRecord<>(2, 1),
+ new StreamRecord<>(2, 2),
+ new StreamRecord<>(2, 3),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(false)
+ .build(),
+ new Watermark(6));
+ }
+
+ @Test
+ public void simpleVariableLengthKeySorting() throws Exception {
Review Comment:
This test the variable length key similar to the test in `SortingDataInputTest`.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1384475401
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/StreamingWithBacklogITCase.java:
##########
@@ -0,0 +1,180 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.streaming.api.datastream;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.AggregateFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.hybrid.HybridSource;
+import org.apache.flink.connector.datagen.source.DataGeneratorSource;
+import org.apache.flink.connector.datagen.source.GeneratorFunction;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL_DURING_BACKLOG;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Integration test for streaming job with backlog. */
+public class StreamingWithBacklogITCase {
+ @Test
+ void testKeyedAggregationWithBacklog() throws Exception {
+ final Configuration config = new Configuration();
+ config.set(CHECKPOINTING_INTERVAL_DURING_BACKLOG, Duration.ZERO);
Review Comment:
The result is not deterministic with parallelism greater than 1. I added two tests with parallelism 1 to test the result.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33202][runtime] Support switching from batch to stream mode to improve throughput when processing backlog data [flink]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1361503174
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/RecordAttributesBuilder.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.streamrecord;
+
+import org.apache.flink.annotation.Experimental;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/** The builder class for {@link RecordAttributes}. */
+@Experimental
+public class RecordAttributesBuilder {
+ private final List<RecordAttributes> lastRecordAttributesOfInputs;
+ @Nullable private Boolean backlog = null;
+
+ /**
+ * This constructor takes a list of the last recordAttributes received from each of the
+ * operator's inputs. When this list is not empty, it will be used to determine the default
+ * values for those attributes that have not been explicitly set by caller.
+ */
+ public RecordAttributesBuilder(List<RecordAttributes> lastRecordAttributesOfInputs) {
+ this.lastRecordAttributesOfInputs = lastRecordAttributesOfInputs;
+ }
+
+ public RecordAttributesBuilder setBacklog(boolean backlog) {
+ this.backlog = backlog;
+ return this;
+ }
+
+ /**
+ * If any operator attribute is null, we will log it at DEBUG level and determine a non-null
+ * default value as described below.
+ *
+ * <p>Default value for backlog: if any element in lastRecordAttributesOfInputs has
+ * backlog=true, use true. Otherwise, use false.
+ */
+ public RecordAttributes build() {
+ if (backlog == null) {
+ backlog = getDefaultBacklog(lastRecordAttributesOfInputs);
Review Comment:
Let's add DEBUG level log according to the JavaDoc.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/RecordAttributesBuilder.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.streamrecord;
+
+import org.apache.flink.annotation.Experimental;
+
+import javax.annotation.Nullable;
+
+import java.util.List;
+
+/** The builder class for {@link RecordAttributes}. */
+@Experimental
+public class RecordAttributesBuilder {
+ private final List<RecordAttributes> lastRecordAttributesOfInputs;
+ @Nullable private Boolean backlog = null;
+
+ /**
+ * This constructor takes a list of the last recordAttributes received from each of the
+ * operator's inputs. When this list is not empty, it will be used to determine the default
+ * values for those attributes that have not been explicitly set by caller.
+ */
+ public RecordAttributesBuilder(List<RecordAttributes> lastRecordAttributesOfInputs) {
+ this.lastRecordAttributesOfInputs = lastRecordAttributesOfInputs;
+ }
+
+ public RecordAttributesBuilder setBacklog(boolean backlog) {
+ this.backlog = backlog;
+ return this;
+ }
+
+ /**
+ * If any operator attribute is null, we will log it at DEBUG level and determine a non-null
+ * default value as described below.
+ *
+ * <p>Default value for backlog: if any element in lastRecordAttributesOfInputs has
+ * backlog=true, use true. Otherwise, use false.
+ */
+ public RecordAttributes build() {
+ if (backlog == null) {
+ backlog = getDefaultBacklog(lastRecordAttributesOfInputs);
+ }
+ return new RecordAttributes(backlog);
+ }
+
+ private boolean getDefaultBacklog(List<RecordAttributes> lastRecordAttributesOfInputs) {
Review Comment:
It might be better to either make this method static or reuse the `lastRecordAttributesOfInputs` variable in the builder class.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/source/event/IsBacklogEvent.java:
##########
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.source.event;
+
+import org.apache.flink.runtime.operators.coordination.OperatorEvent;
+
+/** A source event that notify the source of the backlog status. */
+public class IsBacklogEvent implements OperatorEvent {
Review Comment:
nit: it might be better to rename it as `BacklogEvent`.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java:
##########
@@ -649,4 +659,52 @@ public OperatorID getOperatorID() {
protected Optional<InternalTimeServiceManager<?>> getTimeServiceManager() {
return Optional.ofNullable(timeServiceManager);
}
+
+ @Experimental
+ public void processRecordAttributes(RecordAttributes recordAttributes) throws Exception {
+ lastRecordAttributes1 = recordAttributes;
+ if (timeServiceManager != null
+ && timeServiceManager instanceof InternalBacklogAwareTimerServiceManagerImpl) {
+ final InternalBacklogAwareTimerServiceManagerImpl<?> backlogAwareTimerServiceManager =
+ (InternalBacklogAwareTimerServiceManagerImpl<?>) timeServiceManager;
+ if (recordAttributes instanceof InternalRecordAttributes) {
+ backlogAwareTimerServiceManager.setMaxWatermarkDuringBacklog(
+ ((InternalRecordAttributes) recordAttributes)
+ .getMaxWatermarkDuringBacklog());
+ }
+ backlogAwareTimerServiceManager.setBacklog(recordAttributes.isBacklog());
+ }
+ output.emitRecordAttributes(
+ new RecordAttributesBuilder(Collections.singletonList(recordAttributes)).build());
+ }
+
+ @Experimental
+ public void processRecordAttributes1(RecordAttributes recordAttributes) throws Exception {
+ lastRecordAttributes1 = recordAttributes;
+ List<RecordAttributes> lastRecordAttributes = getTwoInputsLastRecordAttributes();
+ output.emitRecordAttributes(new RecordAttributesBuilder(lastRecordAttributes).build());
+ }
+
+ @Experimental
+ public void processRecordAttributes2(RecordAttributes recordAttributes) throws Exception {
+ lastRecordAttributes2 = recordAttributes;
+ List<RecordAttributes> lastRecordAttributes = getTwoInputsLastRecordAttributes();
+ output.emitRecordAttributes(new RecordAttributesBuilder(lastRecordAttributes).build());
+ }
+
+ private List<RecordAttributes> getTwoInputsLastRecordAttributes() {
+ List<RecordAttributes> lastRecordAttributes;
+ if (lastRecordAttributes1 == null && lastRecordAttributes2 == null) {
+ // should not reach here.
+ throw new RuntimeException(
+ "lastRecordAttributes1 and lastRecordAttributes2 cannot be both null.");
Review Comment:
nit: It might be better to move this condition to the last of these conditions to reduce the average code path.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalBacklogAwareTimerServiceImpl.java:
##########
@@ -0,0 +1,114 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+/**
+ * InternalBacklogAwareTimerServiceImpl uses {@link InternalTimerServiceImpl} for event time timers
+ * during non-backlog processing and uses {@link BacklogTimeService} for event time timers during
+ * backlog processing. All the processing time timers are managed by the {@link
+ * InternalTimerServiceImpl}.
+ */
+@Internal
+public class InternalBacklogAwareTimerServiceImpl<K, N> implements InternalTimerService<N> {
+
+ private final InternalTimerServiceImpl<K, N> realTimeInternalTimeService;
+ private final BacklogTimeService<K, N> backlogTimeService;
+ private InternalTimerService<N> currentInternalTimerService;
+
+ public InternalBacklogAwareTimerServiceImpl(
+ InternalTimerServiceImpl<K, N> realTimeInternalTimeService,
+ BacklogTimeService<K, N> backlogTimeService) {
+ this.realTimeInternalTimeService = realTimeInternalTimeService;
+ this.backlogTimeService = backlogTimeService;
+ this.currentInternalTimerService = realTimeInternalTimeService;
+ }
+
+ @Override
+ public long currentProcessingTime() {
+ return realTimeInternalTimeService.currentProcessingTime();
+ }
+
+ @Override
+ public long currentWatermark() {
+ return currentInternalTimerService.currentWatermark();
+ }
+
+ @Override
+ public void registerProcessingTimeTimer(N namespace, long time) {
+ realTimeInternalTimeService.registerProcessingTimeTimer(namespace, time);
+ }
+
+ @Override
+ public void deleteProcessingTimeTimer(N namespace, long time) {
+ realTimeInternalTimeService.deleteProcessingTimeTimer(namespace, time);
+ }
+
+ @Override
+ public void registerEventTimeTimer(N namespace, long time) {
+ currentInternalTimerService.registerEventTimeTimer(namespace, time);
+ }
+
+ @Override
+ public void deleteEventTimeTimer(N namespace, long time) {
+ currentInternalTimerService.deleteEventTimeTimer(namespace, time);
+ }
+
+ @Override
+ public void forEachEventTimeTimer(BiConsumerWithException<N, Long, Exception> consumer)
+ throws Exception {
+ currentInternalTimerService.forEachEventTimeTimer(consumer);
+ }
+
+ @Override
+ public void forEachProcessingTimeTimer(BiConsumerWithException<N, Long, Exception> consumer)
+ throws Exception {
+ realTimeInternalTimeService.forEachProcessingTimeTimer(consumer);
+ }
+
+ public void advanceWatermark(long timestamp) throws Exception {
+ realTimeInternalTimeService.advanceWatermark(timestamp);
+ }
+
+ public void setMaxWatermarkDuringBacklog(long timestamp) {
+ backlogTimeService.setMaxWatermarkDuringBacklog(timestamp);
+ }
+
+ public void setBacklog(boolean backlog) throws Exception {
+ if (currentInternalTimerService == backlogTimeService && !backlog) {
+ // Switch to non backlog
+ backlogTimeService.setCurrentKey(null);
+ currentInternalTimerService = realTimeInternalTimeService;
+ return;
Review Comment:
It might be better to advance realTimeInternalTimeService's watermark to maxWatermarkDuringBacklog.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalBacklogAwareTimerServiceManagerImpl.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupStatePartitionStreamProvider;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.KeyedStateCheckpointOutputStream;
+import org.apache.flink.runtime.state.PriorityQueueSetFactory;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.WrappingRuntimeException;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * InternalBacklogAwareTimerServiceManagerImpl keeps track of all the {@link
+ * InternalBacklogAwareTimerServiceImpl}.
+ */
+@Internal
+public class InternalBacklogAwareTimerServiceManagerImpl<K>
+ extends InternalTimeServiceManagerImpl<K>
+ implements InternalTimeServiceManager<K>, KeyedStateBackend.KeySelectionListener<K> {
+
+ private final Map<String, InternalBacklogAwareTimerServiceImpl<K, ?>> timerServices =
+ new HashMap<>();
+
+ private boolean backlog = false;
+
+ InternalBacklogAwareTimerServiceManagerImpl(
+ KeyGroupRange localKeyGroupRange,
+ KeyContext keyContext,
+ PriorityQueueSetFactory priorityQueueSetFactory,
+ ProcessingTimeService processingTimeService,
+ StreamTaskCancellationContext cancellationContext) {
+ super(
+ localKeyGroupRange,
+ keyContext,
+ priorityQueueSetFactory,
+ processingTimeService,
+ cancellationContext);
+ }
+
+ @Override
+ public <N> InternalTimerService<N> getInternalTimerService(
+ String name,
+ TypeSerializer<K> keySerializer,
+ TypeSerializer<N> namespaceSerializer,
+ Triggerable<K, N> triggerable) {
+
+ final InternalTimerServiceImpl<K, N> internalTimerService =
+ (InternalTimerServiceImpl<K, N>)
+ super.getInternalTimerService(
+ name, keySerializer, namespaceSerializer, triggerable);
+ final KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> eventTimeTimersQueue =
+ internalTimerService.getEventTimeTimersQueue();
+ final KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>>
+ processingTimeTimersQueue = internalTimerService.getProcessingTimeTimersQueue();
+
+ final BacklogTimeService<K, N> backlogTimeService =
+ new BacklogTimeService<>(
+ processingTimeService,
+ triggerable,
+ eventTimeTimersQueue,
+ processingTimeTimersQueue);
+
+ InternalBacklogAwareTimerServiceImpl<K, N> timerService =
+ (InternalBacklogAwareTimerServiceImpl<K, N>) timerServices.get(name);
+ if (timerService == null) {
Review Comment:
It might be better to move the logics above this line to below this line.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/SortingBacklogDataInput.java:
##########
@@ -0,0 +1,318 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators.sort;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.AlgorithmOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.io.AvailabilityProvider;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.jobgraph.tasks.TaskInvokable;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.sort.ExternalSorter;
+import org.apache.flink.runtime.operators.sort.PushSorter;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.DataInputStatus;
+import org.apache.flink.streaming.runtime.io.StreamTaskInput;
+import org.apache.flink.streaming.runtime.streamrecord.InternalRecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * SortingBacklogDataInput forwards the input records to output util it switch to backlog
+ * processing. During backlog processing, it buffers and sorts the input records and outputs the
+ * sorted records to the downstream when switching to non backlog processing.
+ */
+@Internal
+public class SortingBacklogDataInput<T, K> implements StreamTaskInput<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SortingBacklogDataInput.class);
+
+ private final StreamTaskInput<T> wrappedInput;
+ private final PushSorter<Tuple2<byte[], StreamRecord<T>>> sorter;
+ private final KeySelector<T, K> keySelector;
+ private final TypeSerializer<K> keySerializer;
+ private final DataOutputSerializer dataOutputSerializer;
+ private final SortingDataOutput sortingDataOutput;
+ private final StreamTask.CanEmitBatchOfRecordsChecker canEmitBatchOfRecords;
+ private MutableObjectIterator<Tuple2<byte[], StreamRecord<T>>> sortedInput = null;
+ private long watermarkSeenDuringBacklog = Long.MIN_VALUE;
+
+ private volatile OperatingMode mode = OperatingMode.PROCESSING_REALTIME;
+
+ private enum OperatingMode {
+ PROCESSING_REALTIME,
+ SORTING_BACKLOG,
+ FLUSHING_BACKLOG
Review Comment:
It could help improve readability to add brief JavaDocs to these modes and how they convert to each other.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/BacklogTimeServiceTest.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueSetFactory;
+import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.streaming.api.operators.TimeServiceTestUtils.createBacklogTimerService;
+import static org.apache.flink.streaming.api.operators.TimeServiceTestUtils.createTimerQueue;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link BacklogTimeService}. */
+public class BacklogTimeServiceTest {
+
+ @Test
+ public void testTriggerEventTimeTimer() throws Exception {
+ List<Long> timers = new ArrayList<>();
+ TestProcessingTimeService processingTimeService = new TestProcessingTimeService();
+ KeyGroupRange testKeyGroupRange = new KeyGroupRange(0, 1);
+ final HeapPriorityQueueSetFactory priorityQueueSetFactory =
+ new HeapPriorityQueueSetFactory(testKeyGroupRange, 1, 128);
+
+ final TimerSerializer<Integer, String> timerSerializer =
+ new TimerSerializer<>(IntSerializer.INSTANCE, StringSerializer.INSTANCE);
+ final BacklogTimeService<Integer, String> timeService =
+ createBacklogTimerService(
+ processingTimeService,
+ TestTriggerable.eventTimeTrigger(
+ (timer) -> timers.add(timer.getTimestamp())),
+ createTimerQueue(
+ "processingTimerQueue", timerSerializer, priorityQueueSetFactory),
+ createTimerQueue(
+ "eventTimerQueue", timerSerializer, priorityQueueSetFactory));
+
+ timeService.setCurrentKey(1);
+ timeService.registerEventTimeTimer("a", 0);
+ timeService.registerEventTimeTimer("a", 2);
+ timeService.registerEventTimeTimer("a", 1);
+ timeService.setMaxWatermarkDuringBacklog(2);
Review Comment:
nit: Let's also register a timer whose timestamp is larger than max watermark, and verify that this timer will not be emitted when current key changes.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sort/SortingBacklogDataInputTest.java:
##########
@@ -0,0 +1,235 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators.sort;
+
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.runtime.operators.testutils.DummyInvokable;
+import org.apache.flink.runtime.operators.testutils.MockEnvironment;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.DataInputStatus;
+import org.apache.flink.streaming.runtime.streamrecord.InternalRecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.Collections;
+
+/** Tests for {@link SortingBacklogDataInput}. */
+public class SortingBacklogDataInputTest {
+ @Test
+ public void sortingDuringBacklog() throws Exception {
+ CollectingDataOutput<Integer> collectingDataOutput = new CollectingDataOutput<>();
+ CollectionDataInput<Integer> input =
+ new CollectionDataInput<>(
+ Arrays.asList(
+ new StreamRecord<>(2, 0),
+ new StreamRecord<>(1, 0),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(true)
+ .build(),
+ new StreamRecord<>(1, 3),
+ new StreamRecord<>(1, 1),
+ new StreamRecord<>(2, 1),
+ new StreamRecord<>(2, 3),
+ new StreamRecord<>(1, 2),
+ new StreamRecord<>(2, 2),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(false)
+ .build(),
+ new StreamRecord<>(1, 4),
+ new StreamRecord<>(2, 4),
+ new StreamRecord<>(1, 5)));
+ MockEnvironment environment = MockEnvironment.builder().build();
+ SortingBacklogDataInput<Integer, Integer> sortingDataInput =
+ new SortingBacklogDataInput<>(
+ input,
+ new IntSerializer(),
+ new IntSerializer(),
+ (KeySelector<Integer, Integer>) value -> value,
+ environment.getMemoryManager(),
+ environment.getIOManager(),
+ true,
+ 1.0,
+ new Configuration(),
+ new DummyInvokable(),
+ new ExecutionConfig(),
+ () -> true);
+
+ DataInputStatus inputStatus;
+ do {
+ inputStatus = sortingDataInput.emitNext(collectingDataOutput);
+ } while (inputStatus != DataInputStatus.END_OF_INPUT);
+
+ org.assertj.core.api.Assertions.assertThat(collectingDataOutput.events)
+ .containsExactly(
+ new StreamRecord<>(2, 0),
+ new StreamRecord<>(1, 0),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(true)
+ .build(),
+ new InternalRecordAttributes(true, Long.MIN_VALUE),
+ new StreamRecord<>(1, 1),
+ new StreamRecord<>(1, 2),
+ new StreamRecord<>(1, 3),
+ new StreamRecord<>(2, 1),
+ new StreamRecord<>(2, 2),
+ new StreamRecord<>(2, 3),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(false)
+ .build(),
+ new StreamRecord<>(1, 4),
+ new StreamRecord<>(2, 4),
+ new StreamRecord<>(1, 5));
+ }
+
+ @Test
+ public void watermarkPropagation() throws Exception {
+ CollectingDataOutput<Integer> collectingDataOutput = new CollectingDataOutput<>();
+ CollectionDataInput<Integer> input =
+ new CollectionDataInput<>(
+ Arrays.asList(
+ new StreamRecord<>(1, 3),
+ new Watermark(1),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(true)
+ .build(),
+ new StreamRecord<>(1, 1),
+ new Watermark(2),
+ new StreamRecord<>(2, 1),
+ new Watermark(3),
+ new StreamRecord<>(2, 3),
+ new Watermark(4),
+ new StreamRecord<>(1, 2),
+ new Watermark(5),
+ new StreamRecord<>(2, 2),
+ new Watermark(6),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(false)
+ .build()));
+ MockEnvironment environment = MockEnvironment.builder().build();
+ SortingBacklogDataInput<Integer, Integer> sortingDataInput =
+ new SortingBacklogDataInput<>(
+ input,
+ new IntSerializer(),
+ new IntSerializer(),
+ (KeySelector<Integer, Integer>) value -> value,
+ environment.getMemoryManager(),
+ environment.getIOManager(),
+ true,
+ 1.0,
+ new Configuration(),
+ new DummyInvokable(),
+ new ExecutionConfig(),
+ () -> true);
+
+ DataInputStatus inputStatus;
+ do {
+ inputStatus = sortingDataInput.emitNext(collectingDataOutput);
+ } while (inputStatus != DataInputStatus.END_OF_INPUT);
+
+ org.assertj.core.api.Assertions.assertThat(collectingDataOutput.events)
+ .containsExactly(
+ new StreamRecord<>(1, 3),
+ new Watermark(1),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(true)
+ .build(),
+ new InternalRecordAttributes(true, 6L),
+ new StreamRecord<>(1, 1),
+ new StreamRecord<>(1, 2),
+ new StreamRecord<>(2, 1),
+ new StreamRecord<>(2, 2),
+ new StreamRecord<>(2, 3),
+ new RecordAttributesBuilder(Collections.emptyList())
+ .setBacklog(false)
+ .build(),
+ new Watermark(6));
+ }
+
+ @Test
+ public void simpleVariableLengthKeySorting() throws Exception {
Review Comment:
What does this test case verify in addition to other test cases?
##########
flink-tests/src/test/java/org/apache/flink/test/streaming/api/datastream/StreamingWithBacklogITCase.java:
##########
@@ -0,0 +1,182 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.test.streaming.api.datastream;
+
+import org.apache.flink.api.common.eventtime.WatermarkStrategy;
+import org.apache.flink.api.common.functions.AggregateFunction;
+import org.apache.flink.api.common.functions.ReduceFunction;
+import org.apache.flink.api.common.typeinfo.Types;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.connector.base.source.hybrid.HybridSource;
+import org.apache.flink.connector.datagen.source.DataGeneratorSource;
+import org.apache.flink.connector.datagen.source.GeneratorFunction;
+import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
+import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
+import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
+import org.apache.flink.streaming.api.windowing.time.Time;
+import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.CollectionUtil;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Duration;
+import java.util.List;
+
+import static org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL_DURING_BACKLOG;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Integration test for streaming job with backlog. */
+public class StreamingWithBacklogITCase {
+ @Test
+ void testKeyedAggregationWithBacklog() throws Exception {
+ final Configuration config = new Configuration();
+ config.set(CHECKPOINTING_INTERVAL_DURING_BACKLOG, Duration.ZERO);
+ final StreamExecutionEnvironment env =
+ StreamExecutionEnvironment.getExecutionEnvironment(config);
+ env.setParallelism(1);
Review Comment:
It might be better not to require users to set parallelism to 1 if they want to use this functionality.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/source/event/IsBacklogEvent.java:
##########
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.source.event;
+
+import org.apache.flink.runtime.operators.coordination.OperatorEvent;
+
+/** A source event that notify the source of the backlog status. */
Review Comment:
nit: "the source" -> "the source operator"
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/InternalRecordAttributes.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.streamrecord;
+
+import org.apache.flink.annotation.Internal;
+
+import java.util.Objects;
+
+/** The RecordAttributes that contains extra information to be used internally. */
+@Internal
+public class InternalRecordAttributes extends RecordAttributes {
+ private final long maxWatermarkDuringBacklog;
+
+ public InternalRecordAttributes(boolean backlog, long backlogWatermark) {
Review Comment:
It might be better to rename `backlogWatermark` to `maxWatermarkDuringBacklog`.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/InternalRecordAttributes.java:
##########
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.streamrecord;
+
+import org.apache.flink.annotation.Internal;
+
+import java.util.Objects;
+
+/** The RecordAttributes that contains extra information to be used internally. */
+@Internal
+public class InternalRecordAttributes extends RecordAttributes {
+ private final long maxWatermarkDuringBacklog;
+
+ public InternalRecordAttributes(boolean backlog, long backlogWatermark) {
+ super(backlog);
+ this.maxWatermarkDuringBacklog = backlogWatermark;
+ }
+
+ public long getMaxWatermarkDuringBacklog() {
+ return maxWatermarkDuringBacklog;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ InternalRecordAttributes that = (InternalRecordAttributes) o;
+ return maxWatermarkDuringBacklog == that.maxWatermarkDuringBacklog;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(super.hashCode(), maxWatermarkDuringBacklog);
+ }
+
+ @Override
+ public String toString() {
+ return "InternalRecordAttributes{"
+ + "watermarkDuringBacklog="
+ + maxWatermarkDuringBacklog
Review Comment:
Let's add backlog's value to the string representation as well.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/BacklogTimeService.java:
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.streaming.api.operators.sorted.state.BatchExecutionInternalTimeService;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+
+import java.util.LinkedList;
+import java.util.List;
+
+/**
+ * An implementation of a {@link InternalTimerService} that manages timers with a single active key
+ * at a time. This is used by {@link
+ * org.apache.flink.streaming.api.operators.InternalBacklogAwareTimerServiceImpl} during backlog
+ * processing.
+ */
+@Internal
+public class BacklogTimeService<K, N> extends BatchExecutionInternalTimeService<K, N> {
+ private long maxWatermarkDuringBacklog;
+
+ public BacklogTimeService(
+ ProcessingTimeService processingTimeService,
+ Triggerable<K, N> triggerTarget,
+ KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> eventTimeTimersQueue,
+ KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>>
+ processingTimeTimersQueue) {
+ super(
+ processingTimeService,
+ triggerTarget,
+ eventTimeTimersQueue,
+ processingTimeTimersQueue);
+ }
+
+ @Override
+ public void registerProcessingTimeTimer(N namespace, long time) {
Review Comment:
Let's override `deleteProcessingTimeTimer` as well.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/streamrecord/StreamElementSerializer.java:
##########
Review Comment:
We should also add RecordAttributes serialization process in `StreamElement deserialize(StreamElement reuse, DataInputView source)` and `copy(DataInputView source, DataOutputView target)`.
Let's also add tests for RecordAttributes in `StreamElementSerializerTest`.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/SourceOperator.java:
##########
@@ -569,6 +571,12 @@ public void handleOperatorEvent(OperatorEvent event) {
sourceReader.handleSourceEvents(((SourceEventWrapper) event).getSourceEvent());
} else if (event instanceof NoMoreSplitsEvent) {
sourceReader.notifyNoMoreSplits();
+ } else if (event instanceof IsBacklogEvent) {
+ eventTimeLogic.triggerPeriodicEmit(System.currentTimeMillis());
Review Comment:
Should we only invoke this method when `event.isBacklog() == true`, and cancel the previously scheduled periodicEmit in `eventTimeLogic.startPeriodicWatermarkEmits`?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalBacklogAwareTimerServiceManagerImpl.java:
##########
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupStatePartitionStreamProvider;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.KeyedStateCheckpointOutputStream;
+import org.apache.flink.runtime.state.PriorityQueueSetFactory;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.WrappingRuntimeException;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * InternalBacklogAwareTimerServiceManagerImpl keeps track of all the {@link
+ * InternalBacklogAwareTimerServiceImpl}.
+ */
+@Internal
+public class InternalBacklogAwareTimerServiceManagerImpl<K>
+ extends InternalTimeServiceManagerImpl<K>
+ implements InternalTimeServiceManager<K>, KeyedStateBackend.KeySelectionListener<K> {
+
+ private final Map<String, InternalBacklogAwareTimerServiceImpl<K, ?>> timerServices =
+ new HashMap<>();
+
+ private boolean backlog = false;
+
+ InternalBacklogAwareTimerServiceManagerImpl(
+ KeyGroupRange localKeyGroupRange,
+ KeyContext keyContext,
+ PriorityQueueSetFactory priorityQueueSetFactory,
+ ProcessingTimeService processingTimeService,
+ StreamTaskCancellationContext cancellationContext) {
+ super(
+ localKeyGroupRange,
+ keyContext,
+ priorityQueueSetFactory,
+ processingTimeService,
+ cancellationContext);
+ }
+
+ @Override
+ public <N> InternalTimerService<N> getInternalTimerService(
+ String name,
+ TypeSerializer<K> keySerializer,
+ TypeSerializer<N> namespaceSerializer,
+ Triggerable<K, N> triggerable) {
+
+ final InternalTimerServiceImpl<K, N> internalTimerService =
+ (InternalTimerServiceImpl<K, N>)
+ super.getInternalTimerService(
+ name, keySerializer, namespaceSerializer, triggerable);
+ final KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> eventTimeTimersQueue =
+ internalTimerService.getEventTimeTimersQueue();
+ final KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>>
+ processingTimeTimersQueue = internalTimerService.getProcessingTimeTimersQueue();
+
+ final BacklogTimeService<K, N> backlogTimeService =
+ new BacklogTimeService<>(
+ processingTimeService,
+ triggerable,
+ eventTimeTimersQueue,
+ processingTimeTimersQueue);
Review Comment:
Would it be simpler to pass null or an empty queue as `processingTimeTimersQueue`?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/OperatorAttributesBuilder.java:
##########
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.annotation.Experimental;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/** The builder class for {@link OperatorAttributes}. */
+@Experimental
+public class OperatorAttributesBuilder {
+
+ private static final Logger LOG = LoggerFactory.getLogger(OperatorAttributesBuilder.class);
+
+ @Nullable private Boolean internalSorterSupported = null;
+
+ public OperatorAttributesBuilder() {}
+
+ public OperatorAttributesBuilder setInternalSorterSupported(boolean internalSorterSupported) {
Review Comment:
This method seems unused yet. Let's invoke this method in the corresponding `getOperatorAttributes` method in KeyedCoProcessOperator and IntervalJoinOperator according to the FLIP's description.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamConfig.java:
##########
@@ -795,6 +795,8 @@ public enum InputRequirement {
*/
SORTED,
+ SORTED_DURING_BACKLOG,
Review Comment:
Let's add a JavaDoc for this enum value, like that in `SORTED` and `PASS_THROUGH`.
Besides, it might be better to add newly introduced enum values to the end of a enum class, otherwise the ordinal of enum values after the newly introduced one (PASS_THROUGH -> 1) might be changed (PASS_THROUGH -> 2).
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java:
##########
@@ -649,4 +659,52 @@ public OperatorID getOperatorID() {
protected Optional<InternalTimeServiceManager<?>> getTimeServiceManager() {
return Optional.ofNullable(timeServiceManager);
}
+
+ @Experimental
+ public void processRecordAttributes(RecordAttributes recordAttributes) throws Exception {
+ lastRecordAttributes1 = recordAttributes;
+ if (timeServiceManager != null
+ && timeServiceManager instanceof InternalBacklogAwareTimerServiceManagerImpl) {
+ final InternalBacklogAwareTimerServiceManagerImpl<?> backlogAwareTimerServiceManager =
+ (InternalBacklogAwareTimerServiceManagerImpl<?>) timeServiceManager;
+ if (recordAttributes instanceof InternalRecordAttributes) {
+ backlogAwareTimerServiceManager.setMaxWatermarkDuringBacklog(
+ ((InternalRecordAttributes) recordAttributes)
+ .getMaxWatermarkDuringBacklog());
+ }
+ backlogAwareTimerServiceManager.setBacklog(recordAttributes.isBacklog());
+ }
+ output.emitRecordAttributes(
+ new RecordAttributesBuilder(Collections.singletonList(recordAttributes)).build());
+ }
+
+ @Experimental
+ public void processRecordAttributes1(RecordAttributes recordAttributes) throws Exception {
+ lastRecordAttributes1 = recordAttributes;
+ List<RecordAttributes> lastRecordAttributes = getTwoInputsLastRecordAttributes();
+ output.emitRecordAttributes(new RecordAttributesBuilder(lastRecordAttributes).build());
Review Comment:
Should we invoke methods like `backlogAwareTimerServiceManager.setBacklog`, like in `processRecordAttributes`?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/MultiInputTransformationTranslator.java:
##########
@@ -64,7 +64,8 @@ protected Collection<Integer> translateForBatchInternal(
IntStream.range(0, inputs.size())
.mapToObj(
idx -> {
- if (keySelectors.get(idx) != null) {
+ if (keySelectors.get(idx) != null
+ && !transformation.isInternalSorterSupported()) {
return StreamConfig.InputRequirement.SORTED;
} else {
Review Comment:
Should we support a third condition here?
```
else if (keySelectors.get(idx) != null
&& transformation.isInternalSorterSupported()) {
return StreamConfig.InputRequirement.SORTED_DURING_BACKLOG;
}
```
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/InternalBacklogAwareTimerServiceImplTest.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueSetFactory;
+import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.streaming.api.operators.TimeServiceTestUtils.createBacklogTimerService;
+import static org.apache.flink.streaming.api.operators.TimeServiceTestUtils.createInternalTimerService;
+import static org.apache.flink.streaming.api.operators.TimeServiceTestUtils.createTimerQueue;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link InternalBacklogAwareTimerServiceImpl}. */
+class InternalBacklogAwareTimerServiceImplTest {
Review Comment:
nit: public class
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/SortingBacklogDataInput.java:
##########
@@ -0,0 +1,318 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators.sort;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.AlgorithmOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.io.AvailabilityProvider;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.jobgraph.tasks.TaskInvokable;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.sort.ExternalSorter;
+import org.apache.flink.runtime.operators.sort.PushSorter;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.DataInputStatus;
+import org.apache.flink.streaming.runtime.io.StreamTaskInput;
+import org.apache.flink.streaming.runtime.streamrecord.InternalRecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * SortingBacklogDataInput forwards the input records to output util it switch to backlog
+ * processing. During backlog processing, it buffers and sorts the input records and outputs the
+ * sorted records to the downstream when switching to non backlog processing.
+ */
+@Internal
+public class SortingBacklogDataInput<T, K> implements StreamTaskInput<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SortingBacklogDataInput.class);
+
+ private final StreamTaskInput<T> wrappedInput;
+ private final PushSorter<Tuple2<byte[], StreamRecord<T>>> sorter;
+ private final KeySelector<T, K> keySelector;
+ private final TypeSerializer<K> keySerializer;
+ private final DataOutputSerializer dataOutputSerializer;
+ private final SortingDataOutput sortingDataOutput;
+ private final StreamTask.CanEmitBatchOfRecordsChecker canEmitBatchOfRecords;
+ private MutableObjectIterator<Tuple2<byte[], StreamRecord<T>>> sortedInput = null;
+ private long watermarkSeenDuringBacklog = Long.MIN_VALUE;
+
+ private volatile OperatingMode mode = OperatingMode.PROCESSING_REALTIME;
+
+ private enum OperatingMode {
+ PROCESSING_REALTIME,
+ SORTING_BACKLOG,
+ FLUSHING_BACKLOG
+ }
+
+ public SortingBacklogDataInput(
+ StreamTaskInput<T> wrappedInput,
+ TypeSerializer<T> typeSerializer,
+ TypeSerializer<K> keySerializer,
+ KeySelector<T, K> keySelector,
+ MemoryManager memoryManager,
+ IOManager ioManager,
+ boolean objectReuse,
+ double managedMemoryFraction,
+ Configuration taskManagerConfiguration,
+ TaskInvokable containingTask,
+ ExecutionConfig executionConfig,
+ StreamTask.CanEmitBatchOfRecordsChecker canEmitBatchOfRecords) {
+ try {
+ this.canEmitBatchOfRecords = canEmitBatchOfRecords;
+ this.sortingDataOutput = new SortingDataOutput();
+ this.keySelector = keySelector;
+ this.keySerializer = keySerializer;
+ int keyLength = keySerializer.getLength();
+ final TypeComparator<Tuple2<byte[], StreamRecord<T>>> comparator;
+ if (keyLength > 0) {
+ this.dataOutputSerializer = new DataOutputSerializer(keyLength);
+ comparator = new FixedLengthByteKeyComparator<>(keyLength);
+ } else {
+ this.dataOutputSerializer = new DataOutputSerializer(64);
+ comparator = new VariableLengthByteKeyComparator<>();
+ }
+ KeyAndValueSerializer<T> keyAndValueSerializer =
+ new KeyAndValueSerializer<>(typeSerializer, keyLength);
+ this.wrappedInput = wrappedInput;
+ this.sorter =
+ ExternalSorter.newBuilder(
+ memoryManager,
+ containingTask,
+ keyAndValueSerializer,
+ comparator,
+ executionConfig)
+ .memoryFraction(managedMemoryFraction)
+ .enableSpilling(
+ ioManager,
+ taskManagerConfiguration.get(
+ AlgorithmOptions.SORT_SPILLING_THRESHOLD))
+ .maxNumFileHandles(
+ taskManagerConfiguration.get(AlgorithmOptions.SPILLING_MAX_FAN))
+ .objectReuse(objectReuse)
+ .largeRecords(
+ taskManagerConfiguration.get(
+ AlgorithmOptions.USE_LARGE_RECORDS_HANDLER))
+ .build();
+ } catch (MemoryAllocationException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public int getInputIndex() {
+ return wrappedInput.getInputIndex();
+ }
+
+ @Override
+ public CompletableFuture<Void> prepareSnapshot(
+ ChannelStateWriter channelStateWriter, long checkpointId) throws CheckpointException {
+ if (mode != OperatingMode.PROCESSING_REALTIME) {
+ throw new UnsupportedOperationException(
+ "Checkpoints are not supported during backlog.");
+ }
+ return wrappedInput.prepareSnapshot(channelStateWriter, checkpointId);
+ }
+
+ @Override
+ public void close() throws IOException {
+ IOException ex = null;
+ try {
+ wrappedInput.close();
+ } catch (IOException e) {
+ ex = ExceptionUtils.firstOrSuppressed(e, ex);
+ }
+
+ try {
+ sorter.close();
+ } catch (IOException e) {
+ ex = ExceptionUtils.firstOrSuppressed(e, ex);
+ }
+
+ if (ex != null) {
+ throw ex;
+ }
+ }
+
+ @Override
+ public DataInputStatus emitNext(DataOutput<T> output) throws Exception {
+ LOG.debug("Emit next, current mode: {}", mode);
+ if (sortingDataOutput.innerOutput != output) {
+ sortingDataOutput.innerOutput = output;
+ }
+
+ if (mode == OperatingMode.PROCESSING_REALTIME) {
+ return wrappedInput.emitNext(sortingDataOutput);
+ }
+
+ if (mode == OperatingMode.SORTING_BACKLOG) {
+ return wrappedInput.emitNext(sortingDataOutput);
+ }
+
+ if (mode == OperatingMode.FLUSHING_BACKLOG) {
+ while (true) {
+ final DataInputStatus status = emitNextSortedRecord(output);
+ if (status == DataInputStatus.MORE_AVAILABLE
+ && canEmitBatchOfRecords.check()
+ && mode == OperatingMode.FLUSHING_BACKLOG) {
+ continue;
+ }
+ return status;
+ }
+ }
+
+ // Should never reach here
+ throw new RuntimeException(String.format("Unknown OperatingMode %s", mode));
+ }
+
+ @Nonnull
+ private DataInputStatus emitNextSortedRecord(DataOutput<T> output) throws Exception {
+ Tuple2<byte[], StreamRecord<T>> next = sortedInput.next();
+ if (next != null) {
+ output.emitRecord(next.f1);
+ } else {
+ // Finished flushing
+ mode = OperatingMode.PROCESSING_REALTIME;
+
+ // Send backlog=false downstream
+ output.emitRecordAttributes(
+ new RecordAttributesBuilder(Collections.emptyList()).setBacklog(false).build());
+
+ if (watermarkSeenDuringBacklog > Long.MIN_VALUE) {
+ output.emitWatermark(new Watermark(watermarkSeenDuringBacklog));
+ }
+ }
+ return DataInputStatus.MORE_AVAILABLE;
+ }
+
+ @Override
+ public CompletableFuture<?> getAvailableFuture() {
+ if (mode == OperatingMode.FLUSHING_BACKLOG) {
+ return AvailabilityProvider.AVAILABLE;
+ } else {
+ return wrappedInput.getAvailableFuture();
+ }
+ }
+
+ private class SortingDataOutput implements DataOutput<T> {
+
+ private DataOutput<T> innerOutput;
+
+ @Override
+ public void emitRecord(StreamRecord<T> streamRecord) throws Exception {
+ LOG.debug("Emit record {}", streamRecord.getValue());
+ if (mode == OperatingMode.PROCESSING_REALTIME) {
+ innerOutput.emitRecord(streamRecord);
+ return;
+ }
+
+ if (mode == OperatingMode.SORTING_BACKLOG) {
+ K key = keySelector.getKey(streamRecord.getValue());
+
+ keySerializer.serialize(key, dataOutputSerializer);
+ byte[] serializedKey = dataOutputSerializer.getCopyOfBuffer();
+ dataOutputSerializer.clear();
+
+ sorter.writeRecord(Tuple2.of(serializedKey, streamRecord));
+ return;
+ }
+
+ if (mode == OperatingMode.FLUSHING_BACKLOG) {
+ throw new RuntimeException("Unexpected StreamRecord during FLUSHING_BACKLOG.");
+ }
+ }
+
+ @Override
+ public void emitWatermark(Watermark watermark) throws Exception {
+ if (mode == OperatingMode.PROCESSING_REALTIME) {
+ innerOutput.emitWatermark(watermark);
+ } else {
+ watermarkSeenDuringBacklog =
+ Math.max(watermarkSeenDuringBacklog, watermark.getTimestamp());
+ }
+ }
+
+ @Override
+ public void emitWatermarkStatus(WatermarkStatus watermarkStatus) throws Exception {
+ if (mode == OperatingMode.PROCESSING_REALTIME) {
+ innerOutput.emitWatermarkStatus(watermarkStatus);
+ }
+
+ // Ignore watermark status during backlog
+ }
+
+ @Override
+ public void emitLatencyMarker(LatencyMarker latencyMarker) throws Exception {
+ if (mode == OperatingMode.PROCESSING_REALTIME) {
+ innerOutput.emitLatencyMarker(latencyMarker);
+ }
+
+ // Ignore watermark status during backlog
Review Comment:
nit: ignore latency marker.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/SortingBacklogDataInput.java:
##########
@@ -0,0 +1,318 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators.sort;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.AlgorithmOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.io.AvailabilityProvider;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.jobgraph.tasks.TaskInvokable;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.sort.ExternalSorter;
+import org.apache.flink.runtime.operators.sort.PushSorter;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.DataInputStatus;
+import org.apache.flink.streaming.runtime.io.StreamTaskInput;
+import org.apache.flink.streaming.runtime.streamrecord.InternalRecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * SortingBacklogDataInput forwards the input records to output util it switch to backlog
+ * processing. During backlog processing, it buffers and sorts the input records and outputs the
+ * sorted records to the downstream when switching to non backlog processing.
+ */
+@Internal
+public class SortingBacklogDataInput<T, K> implements StreamTaskInput<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SortingBacklogDataInput.class);
+
+ private final StreamTaskInput<T> wrappedInput;
+ private final PushSorter<Tuple2<byte[], StreamRecord<T>>> sorter;
+ private final KeySelector<T, K> keySelector;
+ private final TypeSerializer<K> keySerializer;
+ private final DataOutputSerializer dataOutputSerializer;
+ private final SortingDataOutput sortingDataOutput;
+ private final StreamTask.CanEmitBatchOfRecordsChecker canEmitBatchOfRecords;
+ private MutableObjectIterator<Tuple2<byte[], StreamRecord<T>>> sortedInput = null;
+ private long watermarkSeenDuringBacklog = Long.MIN_VALUE;
+
+ private volatile OperatingMode mode = OperatingMode.PROCESSING_REALTIME;
+
+ private enum OperatingMode {
+ PROCESSING_REALTIME,
+ SORTING_BACKLOG,
+ FLUSHING_BACKLOG
+ }
+
+ public SortingBacklogDataInput(
+ StreamTaskInput<T> wrappedInput,
+ TypeSerializer<T> typeSerializer,
+ TypeSerializer<K> keySerializer,
+ KeySelector<T, K> keySelector,
+ MemoryManager memoryManager,
+ IOManager ioManager,
+ boolean objectReuse,
+ double managedMemoryFraction,
+ Configuration taskManagerConfiguration,
+ TaskInvokable containingTask,
+ ExecutionConfig executionConfig,
+ StreamTask.CanEmitBatchOfRecordsChecker canEmitBatchOfRecords) {
+ try {
+ this.canEmitBatchOfRecords = canEmitBatchOfRecords;
+ this.sortingDataOutput = new SortingDataOutput();
+ this.keySelector = keySelector;
+ this.keySerializer = keySerializer;
+ int keyLength = keySerializer.getLength();
+ final TypeComparator<Tuple2<byte[], StreamRecord<T>>> comparator;
+ if (keyLength > 0) {
+ this.dataOutputSerializer = new DataOutputSerializer(keyLength);
+ comparator = new FixedLengthByteKeyComparator<>(keyLength);
+ } else {
+ this.dataOutputSerializer = new DataOutputSerializer(64);
+ comparator = new VariableLengthByteKeyComparator<>();
+ }
+ KeyAndValueSerializer<T> keyAndValueSerializer =
+ new KeyAndValueSerializer<>(typeSerializer, keyLength);
+ this.wrappedInput = wrappedInput;
+ this.sorter =
+ ExternalSorter.newBuilder(
+ memoryManager,
+ containingTask,
+ keyAndValueSerializer,
+ comparator,
+ executionConfig)
+ .memoryFraction(managedMemoryFraction)
+ .enableSpilling(
+ ioManager,
+ taskManagerConfiguration.get(
+ AlgorithmOptions.SORT_SPILLING_THRESHOLD))
+ .maxNumFileHandles(
+ taskManagerConfiguration.get(AlgorithmOptions.SPILLING_MAX_FAN))
+ .objectReuse(objectReuse)
+ .largeRecords(
+ taskManagerConfiguration.get(
+ AlgorithmOptions.USE_LARGE_RECORDS_HANDLER))
+ .build();
+ } catch (MemoryAllocationException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public int getInputIndex() {
+ return wrappedInput.getInputIndex();
+ }
+
+ @Override
+ public CompletableFuture<Void> prepareSnapshot(
+ ChannelStateWriter channelStateWriter, long checkpointId) throws CheckpointException {
+ if (mode != OperatingMode.PROCESSING_REALTIME) {
+ throw new UnsupportedOperationException(
+ "Checkpoints are not supported during backlog.");
+ }
+ return wrappedInput.prepareSnapshot(channelStateWriter, checkpointId);
+ }
+
+ @Override
+ public void close() throws IOException {
+ IOException ex = null;
+ try {
+ wrappedInput.close();
+ } catch (IOException e) {
+ ex = ExceptionUtils.firstOrSuppressed(e, ex);
+ }
+
+ try {
+ sorter.close();
+ } catch (IOException e) {
+ ex = ExceptionUtils.firstOrSuppressed(e, ex);
+ }
+
+ if (ex != null) {
+ throw ex;
+ }
+ }
+
+ @Override
+ public DataInputStatus emitNext(DataOutput<T> output) throws Exception {
+ LOG.debug("Emit next, current mode: {}", mode);
+ if (sortingDataOutput.innerOutput != output) {
+ sortingDataOutput.innerOutput = output;
+ }
+
+ if (mode == OperatingMode.PROCESSING_REALTIME) {
+ return wrappedInput.emitNext(sortingDataOutput);
+ }
+
+ if (mode == OperatingMode.SORTING_BACKLOG) {
+ return wrappedInput.emitNext(sortingDataOutput);
+ }
+
+ if (mode == OperatingMode.FLUSHING_BACKLOG) {
+ while (true) {
+ final DataInputStatus status = emitNextSortedRecord(output);
+ if (status == DataInputStatus.MORE_AVAILABLE
+ && canEmitBatchOfRecords.check()
+ && mode == OperatingMode.FLUSHING_BACKLOG) {
+ continue;
+ }
+ return status;
+ }
+ }
+
+ // Should never reach here
+ throw new RuntimeException(String.format("Unknown OperatingMode %s", mode));
+ }
+
+ @Nonnull
+ private DataInputStatus emitNextSortedRecord(DataOutput<T> output) throws Exception {
+ Tuple2<byte[], StreamRecord<T>> next = sortedInput.next();
+ if (next != null) {
+ output.emitRecord(next.f1);
+ } else {
+ // Finished flushing
+ mode = OperatingMode.PROCESSING_REALTIME;
+
+ // Send backlog=false downstream
+ output.emitRecordAttributes(
+ new RecordAttributesBuilder(Collections.emptyList()).setBacklog(false).build());
+
+ if (watermarkSeenDuringBacklog > Long.MIN_VALUE) {
+ output.emitWatermark(new Watermark(watermarkSeenDuringBacklog));
+ }
+ }
+ return DataInputStatus.MORE_AVAILABLE;
+ }
+
+ @Override
+ public CompletableFuture<?> getAvailableFuture() {
+ if (mode == OperatingMode.FLUSHING_BACKLOG) {
+ return AvailabilityProvider.AVAILABLE;
+ } else {
+ return wrappedInput.getAvailableFuture();
+ }
+ }
+
+ private class SortingDataOutput implements DataOutput<T> {
+
+ private DataOutput<T> innerOutput;
+
+ @Override
+ public void emitRecord(StreamRecord<T> streamRecord) throws Exception {
+ LOG.debug("Emit record {}", streamRecord.getValue());
+ if (mode == OperatingMode.PROCESSING_REALTIME) {
+ innerOutput.emitRecord(streamRecord);
+ return;
+ }
+
+ if (mode == OperatingMode.SORTING_BACKLOG) {
+ K key = keySelector.getKey(streamRecord.getValue());
+
+ keySerializer.serialize(key, dataOutputSerializer);
+ byte[] serializedKey = dataOutputSerializer.getCopyOfBuffer();
+ dataOutputSerializer.clear();
+
+ sorter.writeRecord(Tuple2.of(serializedKey, streamRecord));
+ return;
+ }
+
+ if (mode == OperatingMode.FLUSHING_BACKLOG) {
+ throw new RuntimeException("Unexpected StreamRecord during FLUSHING_BACKLOG.");
+ }
Review Comment:
It might be better to throw exception here regardless of OperatingMode value. This helps to handle cases when a new OperatingMode is introduced.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/translators/StreamExecutionUtils.java:
##########
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.translators;
+
+import org.apache.flink.api.dag.Transformation;
+import org.apache.flink.configuration.ExecutionOptions;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.core.memory.ManagedMemoryUseCase;
+import org.apache.flink.streaming.api.graph.StreamConfig;
+import org.apache.flink.streaming.api.graph.StreamNode;
+import org.apache.flink.streaming.api.graph.TransformationTranslator;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL_DURING_BACKLOG;
+
+/** A utility class for applying sorting inputs during backlog processing. */
+public class StreamExecutionUtils {
Review Comment:
The name of this class is more general than the JavaDoc description. It might be better to modify the JavaDoc to avoid mentioning backlog.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/StreamGraphGenerator.java:
##########
@@ -415,6 +417,12 @@ private void configureStreamGraphStreaming(final StreamGraph graph) {
graph.setCheckpointStorage(checkpointStorage);
graph.setSavepointDirectory(savepointDir);
graph.setGlobalStreamExchangeMode(deriveGlobalStreamExchangeModeStreaming());
+
+ if (Duration.ZERO.equals(
+ configuration.get(
+ ExecutionCheckpointingOptions.CHECKPOINTING_INTERVAL_DURING_BACKLOG))) {
+ graph.setTimerServiceProvider(InternalBacklogAwareTimerServiceManagerImpl::create);
+ }
Review Comment:
Why do we need to add a timer service for backlog processing here, while before this PR there is no timer service needed in this method?
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/InternalBacklogAwareTimerServiceImplTest.java:
##########
@@ -0,0 +1,111 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueSetFactory;
+import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.flink.streaming.api.operators.TimeServiceTestUtils.createBacklogTimerService;
+import static org.apache.flink.streaming.api.operators.TimeServiceTestUtils.createInternalTimerService;
+import static org.apache.flink.streaming.api.operators.TimeServiceTestUtils.createTimerQueue;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link InternalBacklogAwareTimerServiceImpl}. */
+class InternalBacklogAwareTimerServiceImplTest {
+
+ @Test
+ void testTriggerEventTimeTimer() throws Exception {
+ KeyGroupRange testKeyGroupRange = new KeyGroupRange(0, 1);
+ List<Tuple2<Integer, Long>> timers = new ArrayList<>();
+ TestProcessingTimeService processingTimeService = new TestProcessingTimeService();
+ final HeapPriorityQueueSetFactory priorityQueueSetFactory =
+ new HeapPriorityQueueSetFactory(testKeyGroupRange, 1, 128);
+ TimerSerializer<Integer, String> timerSerializer =
+ new TimerSerializer<>(IntSerializer.INSTANCE, StringSerializer.INSTANCE);
+ final TestTriggerable<Integer, String> triggerable =
+ TestTriggerable.eventTimeTrigger(
+ (timer) -> timers.add(Tuple2.of(timer.getKey(), timer.getTimestamp())));
+ final KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<Integer, String>>
+ eventTimersQueue =
+ createTimerQueue(
+ "eventTimersQueue", timerSerializer, priorityQueueSetFactory);
+ final KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<Integer, String>>
+ processingTimerQueue =
+ createTimerQueue(
+ "processingTimerQueue", timerSerializer, priorityQueueSetFactory);
+ final BacklogTimeService<Integer, String> backlogTimeService =
+ createBacklogTimerService(
+ processingTimeService, triggerable, processingTimerQueue, eventTimersQueue);
+ final TestKeyContext keyContext = new TestKeyContext();
+ final InternalTimerServiceImpl<Integer, String> internalTimerService =
+ createInternalTimerService(
+ testKeyGroupRange,
+ keyContext,
+ processingTimeService,
+ processingTimerQueue,
+ eventTimersQueue);
+ internalTimerService.startTimerService(
+ IntSerializer.INSTANCE, StringSerializer.INSTANCE, triggerable);
+
+ final InternalBacklogAwareTimerServiceImpl<Integer, String> timerService =
+ new InternalBacklogAwareTimerServiceImpl<>(
+ internalTimerService, backlogTimeService);
+
+ keyContext.setCurrentKey(1);
+ timerService.registerEventTimeTimer("a", 2);
+ timerService.registerEventTimeTimer("a", 1);
+ timerService.registerEventTimeTimer("a", 3);
+ keyContext.setCurrentKey(2);
+ timerService.registerEventTimeTimer("a", 3);
+ timerService.registerEventTimeTimer("a", 1);
+ timerService.advanceWatermark(2);
+ assertThat(timers).containsExactly(Tuple2.of(1, 1L), Tuple2.of(2, 1L), Tuple2.of(1, 2L));
+ timers.clear();
+
+ // switch to backlog processing
+ timerService.setBacklog(true);
+ timerService.setMaxWatermarkDuringBacklog(5);
Review Comment:
`AbstractStreamOperator.processRecordAttributes` would set watermark before set backlog, while the order in this test is reversed. It might be better to keep the same order as in production code.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sort/SortingBacklogDataInput.java:
##########
@@ -0,0 +1,318 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators.sort;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.ExecutionConfig;
+import org.apache.flink.api.common.typeutils.TypeComparator;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.api.java.functions.KeySelector;
+import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.AlgorithmOptions;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.memory.DataOutputSerializer;
+import org.apache.flink.runtime.checkpoint.CheckpointException;
+import org.apache.flink.runtime.checkpoint.channel.ChannelStateWriter;
+import org.apache.flink.runtime.io.AvailabilityProvider;
+import org.apache.flink.runtime.io.disk.iomanager.IOManager;
+import org.apache.flink.runtime.jobgraph.tasks.TaskInvokable;
+import org.apache.flink.runtime.memory.MemoryAllocationException;
+import org.apache.flink.runtime.memory.MemoryManager;
+import org.apache.flink.runtime.operators.sort.ExternalSorter;
+import org.apache.flink.runtime.operators.sort.PushSorter;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.io.DataInputStatus;
+import org.apache.flink.streaming.runtime.io.StreamTaskInput;
+import org.apache.flink.streaming.runtime.streamrecord.InternalRecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.LatencyMarker;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.runtime.tasks.StreamTask;
+import org.apache.flink.streaming.runtime.watermarkstatus.WatermarkStatus;
+import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.util.MutableObjectIterator;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * SortingBacklogDataInput forwards the input records to output util it switch to backlog
+ * processing. During backlog processing, it buffers and sorts the input records and outputs the
+ * sorted records to the downstream when switching to non backlog processing.
+ */
+@Internal
+public class SortingBacklogDataInput<T, K> implements StreamTaskInput<T> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(SortingBacklogDataInput.class);
+
+ private final StreamTaskInput<T> wrappedInput;
+ private final PushSorter<Tuple2<byte[], StreamRecord<T>>> sorter;
+ private final KeySelector<T, K> keySelector;
+ private final TypeSerializer<K> keySerializer;
+ private final DataOutputSerializer dataOutputSerializer;
+ private final SortingDataOutput sortingDataOutput;
+ private final StreamTask.CanEmitBatchOfRecordsChecker canEmitBatchOfRecords;
+ private MutableObjectIterator<Tuple2<byte[], StreamRecord<T>>> sortedInput = null;
+ private long watermarkSeenDuringBacklog = Long.MIN_VALUE;
+
+ private volatile OperatingMode mode = OperatingMode.PROCESSING_REALTIME;
+
+ private enum OperatingMode {
+ PROCESSING_REALTIME,
+ SORTING_BACKLOG,
+ FLUSHING_BACKLOG
+ }
+
+ public SortingBacklogDataInput(
+ StreamTaskInput<T> wrappedInput,
+ TypeSerializer<T> typeSerializer,
+ TypeSerializer<K> keySerializer,
+ KeySelector<T, K> keySelector,
+ MemoryManager memoryManager,
+ IOManager ioManager,
+ boolean objectReuse,
+ double managedMemoryFraction,
+ Configuration taskManagerConfiguration,
+ TaskInvokable containingTask,
+ ExecutionConfig executionConfig,
+ StreamTask.CanEmitBatchOfRecordsChecker canEmitBatchOfRecords) {
+ try {
+ this.canEmitBatchOfRecords = canEmitBatchOfRecords;
+ this.sortingDataOutput = new SortingDataOutput();
+ this.keySelector = keySelector;
+ this.keySerializer = keySerializer;
+ int keyLength = keySerializer.getLength();
+ final TypeComparator<Tuple2<byte[], StreamRecord<T>>> comparator;
+ if (keyLength > 0) {
+ this.dataOutputSerializer = new DataOutputSerializer(keyLength);
+ comparator = new FixedLengthByteKeyComparator<>(keyLength);
+ } else {
+ this.dataOutputSerializer = new DataOutputSerializer(64);
+ comparator = new VariableLengthByteKeyComparator<>();
+ }
+ KeyAndValueSerializer<T> keyAndValueSerializer =
+ new KeyAndValueSerializer<>(typeSerializer, keyLength);
+ this.wrappedInput = wrappedInput;
+ this.sorter =
+ ExternalSorter.newBuilder(
+ memoryManager,
+ containingTask,
+ keyAndValueSerializer,
+ comparator,
+ executionConfig)
+ .memoryFraction(managedMemoryFraction)
+ .enableSpilling(
+ ioManager,
+ taskManagerConfiguration.get(
+ AlgorithmOptions.SORT_SPILLING_THRESHOLD))
+ .maxNumFileHandles(
+ taskManagerConfiguration.get(AlgorithmOptions.SPILLING_MAX_FAN))
+ .objectReuse(objectReuse)
+ .largeRecords(
+ taskManagerConfiguration.get(
+ AlgorithmOptions.USE_LARGE_RECORDS_HANDLER))
+ .build();
+ } catch (MemoryAllocationException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public int getInputIndex() {
+ return wrappedInput.getInputIndex();
+ }
+
+ @Override
+ public CompletableFuture<Void> prepareSnapshot(
+ ChannelStateWriter channelStateWriter, long checkpointId) throws CheckpointException {
+ if (mode != OperatingMode.PROCESSING_REALTIME) {
+ throw new UnsupportedOperationException(
+ "Checkpoints are not supported during backlog.");
+ }
+ return wrappedInput.prepareSnapshot(channelStateWriter, checkpointId);
+ }
+
+ @Override
+ public void close() throws IOException {
+ IOException ex = null;
+ try {
+ wrappedInput.close();
+ } catch (IOException e) {
+ ex = ExceptionUtils.firstOrSuppressed(e, ex);
+ }
+
+ try {
+ sorter.close();
+ } catch (IOException e) {
+ ex = ExceptionUtils.firstOrSuppressed(e, ex);
+ }
+
+ if (ex != null) {
+ throw ex;
+ }
+ }
+
+ @Override
+ public DataInputStatus emitNext(DataOutput<T> output) throws Exception {
+ LOG.debug("Emit next, current mode: {}", mode);
+ if (sortingDataOutput.innerOutput != output) {
+ sortingDataOutput.innerOutput = output;
+ }
+
+ if (mode == OperatingMode.PROCESSING_REALTIME) {
+ return wrappedInput.emitNext(sortingDataOutput);
+ }
+
+ if (mode == OperatingMode.SORTING_BACKLOG) {
+ return wrappedInput.emitNext(sortingDataOutput);
+ }
+
+ if (mode == OperatingMode.FLUSHING_BACKLOG) {
+ while (true) {
+ final DataInputStatus status = emitNextSortedRecord(output);
+ if (status == DataInputStatus.MORE_AVAILABLE
+ && canEmitBatchOfRecords.check()
+ && mode == OperatingMode.FLUSHING_BACKLOG) {
+ continue;
+ }
+ return status;
+ }
+ }
+
+ // Should never reach here
+ throw new RuntimeException(String.format("Unknown OperatingMode %s", mode));
+ }
+
+ @Nonnull
+ private DataInputStatus emitNextSortedRecord(DataOutput<T> output) throws Exception {
+ Tuple2<byte[], StreamRecord<T>> next = sortedInput.next();
+ if (next != null) {
+ output.emitRecord(next.f1);
+ } else {
+ // Finished flushing
+ mode = OperatingMode.PROCESSING_REALTIME;
+
+ // Send backlog=false downstream
+ output.emitRecordAttributes(
+ new RecordAttributesBuilder(Collections.emptyList()).setBacklog(false).build());
+
+ if (watermarkSeenDuringBacklog > Long.MIN_VALUE) {
+ output.emitWatermark(new Watermark(watermarkSeenDuringBacklog));
+ }
+ }
+ return DataInputStatus.MORE_AVAILABLE;
+ }
+
+ @Override
+ public CompletableFuture<?> getAvailableFuture() {
+ if (mode == OperatingMode.FLUSHING_BACKLOG) {
+ return AvailabilityProvider.AVAILABLE;
+ } else {
+ return wrappedInput.getAvailableFuture();
+ }
+ }
+
+ private class SortingDataOutput implements DataOutput<T> {
+
+ private DataOutput<T> innerOutput;
+
+ @Override
+ public void emitRecord(StreamRecord<T> streamRecord) throws Exception {
+ LOG.debug("Emit record {}", streamRecord.getValue());
+ if (mode == OperatingMode.PROCESSING_REALTIME) {
+ innerOutput.emitRecord(streamRecord);
+ return;
+ }
+
+ if (mode == OperatingMode.SORTING_BACKLOG) {
+ K key = keySelector.getKey(streamRecord.getValue());
+
+ keySerializer.serialize(key, dataOutputSerializer);
+ byte[] serializedKey = dataOutputSerializer.getCopyOfBuffer();
+ dataOutputSerializer.clear();
+
+ sorter.writeRecord(Tuple2.of(serializedKey, streamRecord));
+ return;
+ }
+
+ if (mode == OperatingMode.FLUSHING_BACKLOG) {
+ throw new RuntimeException("Unexpected StreamRecord during FLUSHING_BACKLOG.");
+ }
+ }
+
+ @Override
+ public void emitWatermark(Watermark watermark) throws Exception {
+ if (mode == OperatingMode.PROCESSING_REALTIME) {
+ innerOutput.emitWatermark(watermark);
+ } else {
+ watermarkSeenDuringBacklog =
+ Math.max(watermarkSeenDuringBacklog, watermark.getTimestamp());
+ }
+ }
+
+ @Override
+ public void emitWatermarkStatus(WatermarkStatus watermarkStatus) throws Exception {
+ if (mode == OperatingMode.PROCESSING_REALTIME) {
+ innerOutput.emitWatermarkStatus(watermarkStatus);
+ }
+
+ // Ignore watermark status during backlog
+ }
+
+ @Override
+ public void emitLatencyMarker(LatencyMarker latencyMarker) throws Exception {
+ if (mode == OperatingMode.PROCESSING_REALTIME) {
+ innerOutput.emitLatencyMarker(latencyMarker);
+ }
+
+ // Ignore watermark status during backlog
+ }
+
+ @Override
+ public void emitRecordAttributes(RecordAttributes recordAttributes) throws Exception {
+ LOG.debug("Emit record attributes {}", recordAttributes);
+ if (mode == OperatingMode.PROCESSING_REALTIME && recordAttributes.isBacklog()) {
+ // switch to backlog
+ mode = OperatingMode.SORTING_BACKLOG;
+ innerOutput.emitRecordAttributes(recordAttributes);
+ return;
+ }
+
+ if (mode == OperatingMode.SORTING_BACKLOG && !recordAttributes.isBacklog()) {
+ innerOutput.emitRecordAttributes(
+ new InternalRecordAttributes(true, watermarkSeenDuringBacklog));
+ sorter.finishReading();
+ sortedInput = sorter.getIterator();
+ mode = OperatingMode.FLUSHING_BACKLOG;
+ return;
+ }
+
+ if (mode == OperatingMode.FLUSHING_BACKLOG && recordAttributes.isBacklog()) {
+ throw new RuntimeException(
+ "Should not receive record attribute while flushing backlog.");
Review Comment:
The ` && recordAttributes.isBacklog()` in this if condition should be removed according to the error message.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/runtime/io/RecordAttributesValveTest.java:
##########
@@ -0,0 +1,57 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.streaming.api.operators.source.CollectingDataOutput;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link RecordAttributesValve}. */
+class RecordAttributesValveTest {
Review Comment:
nit: public class.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/TestTriggerable.java:
##########
@@ -0,0 +1,60 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.assertj.core.api.Assertions;
+
+import java.util.function.Consumer;
+
+/** Triggerable for test. */
+class TestTriggerable<K, N> implements Triggerable<K, N> {
+
+ private final Consumer<InternalTimer<K, N>> eventTimeHandler;
+ private final Consumer<InternalTimer<K, N>> processingTimeHandler;
+
+ public static <K, N> TestTriggerable<K, N> eventTimeTrigger(
+ Consumer<InternalTimer<K, N>> eventTimeHandler) {
+ return new TestTriggerable<>(
+ eventTimeHandler,
+ timer -> Assertions.fail("We did not expect processing timer to be triggered."));
+ }
+
+ public static <K, N> TestTriggerable<K, N> processingTimeTrigger(
Review Comment:
This method can be removed.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1387397501
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/RecordAttributesValve.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput.DataOutput;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+
+/**
+ * RecordAttributesValve combine RecordAttributes from different input channels. If any of the input
+ * channels is non backlog, the combined RecordAttributes is non backlog.
+ *
+ * <p>Currently, we only support switching the backlog status from null to backlog and backlog to
+ * non-backlog. Switching from non-backlog to backlog is not support at the moment, and it will be
+ * ignored.
Review Comment:
It might be better to throw exception instead of ignoring illegal switches.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/source/coordinator/SourceCoordinatorContext.java:
##########
@@ -629,6 +641,10 @@ private void sendCachedSplitsToNewlyRegisteredReader(int subtaskIndex, int attem
}
}
+ public Boolean isBacklog() {
Review Comment:
It might be better to improve the naming of this method and add JavaDoc explaining when a `null` would be returned.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalBacklogAwareTimerServiceManagerImpl.java:
##########
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.api.operators;
+
+import org.apache.flink.annotation.Internal;
+import org.apache.flink.api.common.typeutils.TypeSerializer;
+import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupStatePartitionStreamProvider;
+import org.apache.flink.runtime.state.KeyedStateBackend;
+import org.apache.flink.runtime.state.KeyedStateCheckpointOutputStream;
+import org.apache.flink.runtime.state.PriorityQueueSetFactory;
+import org.apache.flink.streaming.api.watermark.Watermark;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.util.Preconditions;
+import org.apache.flink.util.WrappingRuntimeException;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * InternalBacklogAwareTimerServiceManagerImpl keeps track of all the {@link
+ * InternalBacklogAwareTimerServiceImpl}.
+ */
+@Internal
+public class InternalBacklogAwareTimerServiceManagerImpl<K>
+ extends InternalTimeServiceManagerImpl<K>
+ implements InternalTimeServiceManager<K>, KeyedStateBackend.KeySelectionListener<K> {
+
+ private final Map<String, InternalBacklogAwareTimerServiceImpl<K, ?>> timerServices =
+ new HashMap<>();
+
+ private boolean backlog = false;
+
+ InternalBacklogAwareTimerServiceManagerImpl(
+ KeyGroupRange localKeyGroupRange,
+ KeyContext keyContext,
+ PriorityQueueSetFactory priorityQueueSetFactory,
+ ProcessingTimeService processingTimeService,
+ StreamTaskCancellationContext cancellationContext) {
+ super(
+ localKeyGroupRange,
+ keyContext,
+ priorityQueueSetFactory,
+ processingTimeService,
+ cancellationContext);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <N> InternalTimerService<N> getInternalTimerService(
+ String name,
+ TypeSerializer<K> keySerializer,
+ TypeSerializer<N> namespaceSerializer,
+ Triggerable<K, N> triggerable) {
+ InternalBacklogAwareTimerServiceImpl<K, N> timerService =
+ (InternalBacklogAwareTimerServiceImpl<K, N>) timerServices.get(name);
+ if (timerService == null) {
+ final InternalTimerServiceImpl<K, N> internalTimerService =
+ (InternalTimerServiceImpl<K, N>)
+ super.getInternalTimerService(
+ name, keySerializer, namespaceSerializer, triggerable);
+ final BacklogTimeService<K, N> backlogTimeService =
+ new BacklogTimeService<>(
+ processingTimeService,
+ triggerable,
+ internalTimerService.getEventTimeTimersQueue());
+ timerService =
+ new InternalBacklogAwareTimerServiceImpl<>(
+ internalTimerService, backlogTimeService);
+ timerServices.put(name, timerService);
+ }
+ return timerService;
+ }
+
+ @Override
+ public void advanceWatermark(Watermark watermark) throws Exception {
+ for (InternalBacklogAwareTimerServiceImpl<?, ?> service : timerServices.values()) {
+ service.advanceWatermark(watermark.getTimestamp());
+ }
+ }
+
+ @Override
+ public void snapshotToRawKeyedState(
+ KeyedStateCheckpointOutputStream stateCheckpointOutputStream, String operatorName)
+ throws Exception {
+ Preconditions.checkState(!backlog, "Cannot snapshot state during backlog.");
+ super.snapshotToRawKeyedState(stateCheckpointOutputStream, operatorName);
+ }
+
+ public static <K> InternalBacklogAwareTimerServiceManagerImpl<K> create(
+ CheckpointableKeyedStateBackend<K> keyedStateBackend,
+ ClassLoader userClassloader,
+ KeyContext keyContext,
+ ProcessingTimeService processingTimeService,
+ Iterable<KeyGroupStatePartitionStreamProvider> rawKeyedStates,
+ StreamTaskCancellationContext cancellationContext)
+ throws Exception {
+
+ final InternalBacklogAwareTimerServiceManagerImpl<K> manager =
+ new InternalBacklogAwareTimerServiceManagerImpl<>(
+ keyedStateBackend.getKeyGroupRange(),
+ keyContext,
+ keyedStateBackend,
+ processingTimeService,
+ cancellationContext);
+
+ keyedStateBackend.registerKeySelectionListener(manager);
+
+ return manager;
+ }
+
+ @Override
+ public void keySelected(K newKey) {
+ try {
+ for (InternalBacklogAwareTimerServiceImpl<K, ?> value : timerServices.values()) {
+ value.setCurrentKey(newKey);
+ }
+ } catch (Exception e) {
+ throw new WrappingRuntimeException(e);
+ }
+ }
+
+ /**
+ * Set the maximum watermark during backlog of all the managed {@link
+ * InternalBacklogAwareTimerServiceImpl}.
+ */
+ public void setMaxWatermarkDuringBacklog(long timestamp) {
+ for (InternalBacklogAwareTimerServiceImpl<K, ?> timerService : timerServices.values()) {
+ timerService.setMaxWatermarkDuringBacklog(timestamp);
+ }
+ }
+
+ /** Set the backlog status all the managed {@link InternalBacklogAwareTimerServiceImpl}. */
+ public void setBacklog(boolean backlog) {
+ try {
+ for (InternalBacklogAwareTimerServiceImpl<K, ?> value : timerServices.values()) {
Review Comment:
nit: might be better to add `if (this.backlog == backlog) return;`.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/io/RecordAttributesValve.java:
##########
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.streaming.runtime.io;
+
+import org.apache.flink.streaming.runtime.io.PushingAsyncDataInput.DataOutput;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributes;
+import org.apache.flink.streaming.runtime.streamrecord.RecordAttributesBuilder;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collections;
+
+/**
+ * RecordAttributesValve combine RecordAttributes from different input channels. If any of the input
+ * channels is non backlog, the combined RecordAttributes is non backlog.
+ *
+ * <p>Currently, we only support switching the backlog status from null to backlog and backlog to
+ * non-backlog. Switching from non-backlog to backlog is not support at the moment, and it will be
+ * ignored.
+ */
+public class RecordAttributesValve {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RecordAttributesValve.class);
+
+ private final int numInputChannels;
+ private final RecordAttributes[] allChannelRecordAttributes;
+
+ private int backlogChannelsCnt = 0;
+ private RecordAttributes lastOutputAttributes = null;
+ private boolean hasBeenBacklog;
Review Comment:
this variable is not used.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33202][runtime] Support switching from batch to stream mode to improve throughput when processing backlog data [flink]
Posted by "Sxnan (via GitHub)" <gi...@apache.org>.
Sxnan commented on code in PR #23521:
URL: https://github.com/apache/flink/pull/23521#discussion_r1366597779
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java:
##########
@@ -649,4 +659,52 @@ public OperatorID getOperatorID() {
protected Optional<InternalTimeServiceManager<?>> getTimeServiceManager() {
return Optional.ofNullable(timeServiceManager);
}
+
+ @Experimental
+ public void processRecordAttributes(RecordAttributes recordAttributes) throws Exception {
+ lastRecordAttributes1 = recordAttributes;
+ if (timeServiceManager != null
+ && timeServiceManager instanceof InternalBacklogAwareTimerServiceManagerImpl) {
+ final InternalBacklogAwareTimerServiceManagerImpl<?> backlogAwareTimerServiceManager =
+ (InternalBacklogAwareTimerServiceManagerImpl<?>) timeServiceManager;
+ if (recordAttributes instanceof InternalRecordAttributes) {
+ backlogAwareTimerServiceManager.setMaxWatermarkDuringBacklog(
+ ((InternalRecordAttributes) recordAttributes)
+ .getMaxWatermarkDuringBacklog());
+ }
+ backlogAwareTimerServiceManager.setBacklog(recordAttributes.isBacklog());
+ }
+ output.emitRecordAttributes(
+ new RecordAttributesBuilder(Collections.singletonList(recordAttributes)).build());
+ }
+
+ @Experimental
+ public void processRecordAttributes1(RecordAttributes recordAttributes) throws Exception {
+ lastRecordAttributes1 = recordAttributes;
+ List<RecordAttributes> lastRecordAttributes = getTwoInputsLastRecordAttributes();
+ output.emitRecordAttributes(new RecordAttributesBuilder(lastRecordAttributes).build());
Review Comment:
`BacklogAwareTimerSerivceManager` is only used by one input StreamOperator.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-33398][runtime] Support switching from batch to stream mode for one input stream operator [flink]
Posted by "SmirAlex (via GitHub)" <gi...@apache.org>.
SmirAlex commented on PR #23521:
URL: https://github.com/apache/flink/pull/23521#issuecomment-1812051645
Hi @Sxnan! Sorry, I understand that I'm not a reviewer, but it happened that I was testing functionality from this MR recently, and I found a bug (in my opinion). It concerns the logic of `RecordAttributesValve`.
I my test there was very little backlog data at source + it had parallelism 4 (actually, any value > 1 suits the case). Due to very short backlog phase, time interval between sending RecordAttributes(isBacklog=true) and RecordAttributes(isBacklog=false) was also very short. In addition, due to the high parallelism one source subtask could send RecordAttributes(isBacklog=false) even before RecordAttributes(isBacklog=true) of another subtask. As a result, race condition have occurred in `RecordAttributesValve#inputRecordAttributes`. `backlogChannelsCnt` was incrementing and decrementing simultaneously, which led to not reaching 'numInputChannels' at all, so no `RecordAttributes` was emitted from `RecordAttributesValve`.
--
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-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org