You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "fredia (via GitHub)" <gi...@apache.org> on 2024/04/16 12:37:45 UTC
[PR] [FLINK-35028][runtime] Timer firing under async execution model [flink]
fredia opened a new pull request, #24672:
URL: https://github.com/apache/flink/pull/24672
<!--
*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
This PR adapts timer firing to async state execution. The entrances are `AbstractAsyncStateStreamOperator ` and `AbstractAsyncStateStreamOperatorV2`, the hot path without this feature is not affected.
## Brief change log
- Introduce `InternalTimerServiceAsyncImpl`.
- Add `InternalTimeServiceManager#getAsyncInternalTimerService` method.
- Override `getInternalTimerService` of `AbstractAsyncStateStreamOperator ` and `AbstractAsyncStateStreamOperatorV2`
## Verifying this change
This change added tests and can be verified as follows:
- `InternalTimerServiceAsyncImplTest`
- `AbstractAsyncStateStreamOperatorTest#testTimerServiceIsAsync`
- - `AbstractAsyncStateStreamOperatorV2Test#testTimerServiceIsAsync`
## Does this pull request potentially affect one of the following parts:
- Dependencies (does it add or upgrade a dependency): (no)
- The public API, i.e., is any changed class annotated with `@Public(Evolving)`: (no)
- The serializers: (no)
- The runtime per-record code paths (performance sensitive): (no)
- Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
- The S3 file system connector: (no)
## Documentation
- Does this pull request introduce a new feature? (yes)
- If yes, how is the feature documented? (JavaDocs)
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24672:
URL: https://github.com/apache/flink/pull/24672#issuecomment-2059006200
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "6507dc0d38dec17e91fa3f722cd924ca82197622",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "6507dc0d38dec17e91fa3f722cd924ca82197622",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* 6507dc0d38dec17e91fa3f722cd924ca82197622 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-35028][runtime] Timer firing under async execution model [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #24672:
URL: https://github.com/apache/flink/pull/24672#discussion_r1574062845
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceAsyncImpl.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.asyncprocessing.AsyncExecutionController;
+import org.apache.flink.runtime.asyncprocessing.RecordContext;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import static org.apache.flink.runtime.asyncprocessing.KeyAccountingUnit.EMPTY_RECORD;
+
+/**
+ * An implementation of {@link InternalTimerService} that is used by {@link
+ * org.apache.flink.streaming.runtime.operators.asyncprocessing.AbstractAsyncStateStreamOperator}.
+ * The timer service will set {@link RecordContext} for the timers before invoking action to
+ * preserve the execution order between timer firing and records processing.
+ *
+ * @see <a
+ * href=https://cwiki.apache.org/confluence/display/FLINK/FLIP-425%3A+Asynchronous+Execution+Model#FLIP425:AsynchronousExecutionModel-Timers>FLIP-425
+ * timers section.</a>
+ * @param <K> Type of timer's key.
+ * @param <N> Type of the namespace to which timers are scoped.
+ */
+public class InternalTimerServiceAsyncImpl<K, N> extends InternalTimerServiceImpl<K, N> {
+
+ private AsyncExecutionController<K> asyncExecutionController;
+
+ InternalTimerServiceAsyncImpl(
+ TaskIOMetricGroup taskIOMetricGroup,
+ KeyGroupRange localKeyGroupRange,
+ KeyContext keyContext,
+ ProcessingTimeService processingTimeService,
+ KeyGroupedInternalPriorityQueue processingTimeTimersQueue,
+ KeyGroupedInternalPriorityQueue eventTimeTimersQueue,
+ StreamTaskCancellationContext cancellationContext,
+ AsyncExecutionController<K> asyncExecutionController) {
+ super(
+ taskIOMetricGroup,
+ localKeyGroupRange,
+ keyContext,
+ processingTimeService,
+ processingTimeTimersQueue,
+ eventTimeTimersQueue,
+ cancellationContext);
+ this.asyncExecutionController = asyncExecutionController;
+ this.processingTimeCallback = this::onProcessingTime;
+ }
+
+ private void onProcessingTime(long time) throws Exception {
+ // null out the timer in case the Triggerable calls registerProcessingTimeTimer()
+ // inside the callback.
+ nextTimer = null;
+
+ InternalTimer<K, N> timer;
+
+ while ((timer = processingTimeTimersQueue.peek()) != null
+ && timer.getTimestamp() <= time
+ && !cancellationContext.isCancelled()) {
+ RecordContext<K> recordCtx =
+ asyncExecutionController.buildContext(EMPTY_RECORD, timer.getKey());
+ recordCtx.retain();
+ asyncExecutionController.setCurrentContext(recordCtx);
+ keyContext.setCurrentKey(timer.getKey());
+ processingTimeTimersQueue.poll();
+ final InternalTimer<K, N> timerToTrigger = timer;
+ asyncExecutionController.syncPointRequestWithCallback(
+ () -> triggerTarget.onProcessingTime(timerToTrigger));
+ taskIOMetricGroup.getNumFiredTimers().inc();
+ recordCtx.release();
+ }
+
+ if (timer != null && nextTimer == null) {
+ nextTimer =
+ processingTimeService.registerTimer(
+ timer.getTimestamp(), this::onProcessingTime);
+ }
+ }
+
+ /**
+ * Advance one watermark, this will fire some event timers.
+ *
+ * @param time the time in watermark.
+ */
+ @Override
+ public void advanceWatermark(long time) throws Exception {
+ currentWatermark = time;
+
+ InternalTimer<K, N> timer;
+
+ while ((timer = eventTimeTimersQueue.peek()) != null
+ && timer.getTimestamp() <= time
+ && !cancellationContext.isCancelled()) {
+ RecordContext<K> recordCtx =
+ asyncExecutionController.buildContext(EMPTY_RECORD, timer.getKey());
+ recordCtx.retain();
+ asyncExecutionController.setCurrentContext(recordCtx);
+ keyContext.setCurrentKey(timer.getKey());
+ eventTimeTimersQueue.poll();
+ final InternalTimer<K, N> timerToTrigger = timer;
+ asyncExecutionController.syncPointRequestWithCallback(
+ () -> triggerTarget.onEventTime(timerToTrigger));
+ taskIOMetricGroup.getNumFiredTimers().inc();
+ recordCtx.release();
+ }
+ }
+
+ protected void foreachTimer(
Review Comment:
Agreed, let `foreachTimer` throw `UnsupportedOperationException` 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-35028][runtime] Timer firing under async execution model [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #24672:
URL: https://github.com/apache/flink/pull/24672#discussion_r1568752322
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/InternalTimerServiceAsyncImplTest.java:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.runtime.asyncprocessing.AsyncExecutionController;
+import org.apache.flink.runtime.asyncprocessing.RecordContext;
+import org.apache.flink.runtime.asyncprocessing.StateExecutor;
+import org.apache.flink.runtime.asyncprocessing.StateRequest;
+import org.apache.flink.runtime.asyncprocessing.StateRequestType;
+import org.apache.flink.runtime.mailbox.SyncMailboxExecutor;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.PriorityQueueSetFactory;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueSetFactory;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link InternalTimerServiceAsyncImpl}. */
+class InternalTimerServiceAsyncImplTest {
Review Comment:
Only the logic of `onProcessingTime` and `advanceWatermark` are changed, the logic of `register/delete` timer or `snapshot/restore` timer are same as `InternalTimerServiceImpl`, I think the `InternalTimerServiceAsyncImplTest` is enough, I tend to consider whether to inherit some tests when there are new changes in the future.
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #24672:
URL: https://github.com/apache/flink/pull/24672#discussion_r1572300164
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/asyncprocessing/AbstractAsyncStateStreamOperatorV2.java:
##########
@@ -113,6 +120,26 @@ public final <T> ThrowingConsumer<StreamRecord<T>, Exception> getRecordProcessor
+ " since this part is handled by the Input.");
}
+ @Override
+ @SuppressWarnings("unchecked")
+ public <K, N> InternalTimerService<N> getInternalTimerService(
+ String name, TypeSerializer<N> namespaceSerializer, Triggerable<K, N> triggerable) {
Review Comment:
A fallback plan when isAsyncStateProcessingEnabled=false might be needed here as well.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceAsyncImpl.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.asyncprocessing.AsyncExecutionController;
+import org.apache.flink.runtime.asyncprocessing.RecordContext;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import static org.apache.flink.runtime.asyncprocessing.KeyAccountingUnit.EMPTY_RECORD;
+
+/**
+ * An implementation of {@link InternalTimerService} that is used by {@link
+ * org.apache.flink.streaming.runtime.operators.asyncprocessing.AbstractAsyncStateStreamOperator}.
+ * The timer service will set {@link RecordContext} for the timers before invoking action to
+ * preserve the execution order between timer firing and records processing.
+ *
+ * @see <a
+ * href=https://cwiki.apache.org/confluence/display/FLINK/FLIP-425%3A+Asynchronous+Execution+Model#FLIP425:AsynchronousExecutionModel-Timers>FLIP-425
+ * timers section.</a>
+ * @param <K> Type of timer's key.
+ * @param <N> Type of the namespace to which timers are scoped.
+ */
+public class InternalTimerServiceAsyncImpl<K, N> extends InternalTimerServiceImpl<K, N> {
+
+ private AsyncExecutionController<K> asyncExecutionController;
+
+ InternalTimerServiceAsyncImpl(
+ TaskIOMetricGroup taskIOMetricGroup,
+ KeyGroupRange localKeyGroupRange,
+ KeyContext keyContext,
+ ProcessingTimeService processingTimeService,
+ KeyGroupedInternalPriorityQueue processingTimeTimersQueue,
+ KeyGroupedInternalPriorityQueue eventTimeTimersQueue,
+ StreamTaskCancellationContext cancellationContext,
+ AsyncExecutionController<K> asyncExecutionController) {
+ super(
+ taskIOMetricGroup,
+ localKeyGroupRange,
+ keyContext,
+ processingTimeService,
+ processingTimeTimersQueue,
+ eventTimeTimersQueue,
+ cancellationContext);
+ this.asyncExecutionController = asyncExecutionController;
+ this.processingTimeCallback = this::onProcessingTime;
+ }
+
+ private void onProcessingTime(long time) throws Exception {
+ // null out the timer in case the Triggerable calls registerProcessingTimeTimer()
+ // inside the callback.
+ nextTimer = null;
+
+ InternalTimer<K, N> timer;
+
+ while ((timer = processingTimeTimersQueue.peek()) != null
+ && timer.getTimestamp() <= time
+ && !cancellationContext.isCancelled()) {
+ RecordContext<K> recordCtx =
+ asyncExecutionController.buildContext(EMPTY_RECORD, timer.getKey());
+ recordCtx.retain();
+ asyncExecutionController.setCurrentContext(recordCtx);
+ keyContext.setCurrentKey(timer.getKey());
+ processingTimeTimersQueue.poll();
+ final InternalTimer<K, N> timerToTrigger = timer;
+ asyncExecutionController.syncPointRequestWithCallback(
+ () -> triggerTarget.onProcessingTime(timerToTrigger));
+ taskIOMetricGroup.getNumFiredTimers().inc();
+ recordCtx.release();
+ }
+
+ if (timer != null && nextTimer == null) {
+ nextTimer =
+ processingTimeService.registerTimer(
+ timer.getTimestamp(), this::onProcessingTime);
+ }
+ }
+
+ /**
+ * Advance one watermark, this will fire some event timers.
+ *
+ * @param time the time in watermark.
+ */
+ @Override
+ public void advanceWatermark(long time) throws Exception {
+ currentWatermark = time;
+
+ InternalTimer<K, N> timer;
+
+ while ((timer = eventTimeTimersQueue.peek()) != null
+ && timer.getTimestamp() <= time
+ && !cancellationContext.isCancelled()) {
+ RecordContext<K> recordCtx =
+ asyncExecutionController.buildContext(EMPTY_RECORD, timer.getKey());
+ recordCtx.retain();
+ asyncExecutionController.setCurrentContext(recordCtx);
+ keyContext.setCurrentKey(timer.getKey());
+ eventTimeTimersQueue.poll();
+ final InternalTimer<K, N> timerToTrigger = timer;
+ asyncExecutionController.syncPointRequestWithCallback(
+ () -> triggerTarget.onEventTime(timerToTrigger));
+ taskIOMetricGroup.getNumFiredTimers().inc();
+ recordCtx.release();
+ }
+ }
+
+ protected void foreachTimer(
Review Comment:
How about throw UnsupportedException for now and add actual implementations later after the evaluation is complete? It might be better not to support certain functions than to support risky functions.
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on PR #24672:
URL: https://github.com/apache/flink/pull/24672#issuecomment-2069427265
Thanks all for the detailed review, merged
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #24672:
URL: https://github.com/apache/flink/pull/24672#discussion_r1570100169
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/KeyAccountingUnit.java:
##########
@@ -32,6 +32,8 @@
*/
public class KeyAccountingUnit<K> {
+ public static final Object EMPTY_RECORD = "EMPTY_RECORD";
Review Comment:
The implementation of `KeyAccountingUnit ` restricts the value(record) from being null.
I changed the `EMPTY_RECORD` to `new Object()`.
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on PR #24672:
URL: https://github.com/apache/flink/pull/24672#issuecomment-2063204411
@Zakelly @yunfengzhou-hub Thanks for the detailed review, I updated the PR and addressed some comments, would you please take a look again?
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #24672:
URL: https://github.com/apache/flink/pull/24672#discussion_r1570075307
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java:
##########
@@ -117,6 +120,7 @@ public class InternalTimerServiceImpl<K, N> implements InternalTimerService<N> {
startIdx = Math.min(keyGroupIdx, startIdx);
}
this.localKeyGroupRangeStartIdx = startIdx;
+ this.processingTimeCallback = this::onProcessingTime;
Review Comment:
Good suggestion👍, override `onProcessingTime` and delete `processingTimeCallback` 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-35028][runtime] Timer firing under async execution model [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24672:
URL: https://github.com/apache/flink/pull/24672#discussion_r1569904139
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/KeyAccountingUnit.java:
##########
@@ -32,6 +32,8 @@
*/
public class KeyAccountingUnit<K> {
+ public static final Object EMPTY_RECORD = "EMPTY_RECORD";
Review Comment:
IMO, this should be under `RecordContext` instead of `KeyAccountingUnit`. And how about using `aec.buildContext(null, key)` everywhere instead of import the EMPTY_RECORD everywhere?
Further more, it might be better the `EMPTY_RECORD` is `new Object()` to avoid coincidence that happens to be the same as the user input.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceAsyncImpl.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.asyncprocessing.AsyncExecutionController;
+import org.apache.flink.runtime.asyncprocessing.RecordContext;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import static org.apache.flink.runtime.asyncprocessing.KeyAccountingUnit.EMPTY_RECORD;
+
+/**
+ * An implementation of {@link InternalTimerService} that is used by {@link
+ * org.apache.flink.streaming.runtime.operators.asyncprocessing.AbstractAsyncStateStreamOperator}.
+ * The timer service will set {@link RecordContext} for the timers before invoking action to
+ * preserve the execution order between timer firing and records processing.
+ *
+ * @see <a
+ * href=https://cwiki.apache.org/confluence/display/FLINK/FLIP-425%3A+Asynchronous+Execution+Model#FLIP425:AsynchronousExecutionModel-Timers>FLIP-425
+ * timers section.</a>
+ * @param <K> Type of timer's key.
+ * @param <N> Type of the namespace to which timers are scoped.
+ */
+public class InternalTimerServiceAsyncImpl<K, N> extends InternalTimerServiceImpl<K, N> {
+
+ private AsyncExecutionController<K> asyncExecutionController;
+
+ InternalTimerServiceAsyncImpl(
+ TaskIOMetricGroup taskIOMetricGroup,
+ KeyGroupRange localKeyGroupRange,
+ KeyContext keyContext,
+ ProcessingTimeService processingTimeService,
+ KeyGroupedInternalPriorityQueue processingTimeTimersQueue,
+ KeyGroupedInternalPriorityQueue eventTimeTimersQueue,
Review Comment:
How about adding `<TimerHeapInternalTimer<K, N>>` as type parameter here?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/asyncprocessing/AbstractAsyncStateStreamOperator.java:
##########
@@ -133,6 +140,48 @@ public final <T> ThrowingConsumer<StreamRecord<T>, Exception> getRecordProcessor
getClass().getName(), inputId));
}
+ /**
+ * Returns a {@link InternalTimerService} that can be used to query current processing time and
+ * event time and to set timers. An operator can have several timer services, where each has its
+ * own namespace serializer. Timer services are differentiated by the string key that is given
+ * when requesting them, if you call this method with the same key multiple times you will get
+ * the same timer service instance in subsequent requests.
+ *
+ * <p>Timers are always scoped to a key, the currently active key of a keyed stream operation.
+ * When a timer fires, this key will also be set as the currently active key.
+ *
+ * <p>Each timer has attached metadata, the namespace. Different timer services can have a
+ * different namespace type. If you don't need namespace differentiation you can use {@link
+ * org.apache.flink.runtime.state.VoidNamespaceSerializer} as the namespace serializer.
+ *
+ * @param name The name of the requested timer service. If no service exists under the given
+ * name a new one will be created and returned.
+ * @param namespaceSerializer {@code TypeSerializer} for the timer namespace.
+ * @param triggerable The {@link Triggerable} that should be invoked when timers fire
+ * @param <N> The type of the timer namespace.
+ */
+ @Override
+ @SuppressWarnings("unchecked")
+ public <K, N> InternalTimerService<N> getInternalTimerService(
+ String name, TypeSerializer<N> namespaceSerializer, Triggerable<K, N> triggerable) {
Review Comment:
Check `isAsyncStateProcessingEnabled()` first and provide fallback?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java:
##########
@@ -117,6 +120,7 @@ public class InternalTimerServiceImpl<K, N> implements InternalTimerService<N> {
startIdx = Math.min(keyGroupIdx, startIdx);
}
this.localKeyGroupRangeStartIdx = startIdx;
+ this.processingTimeCallback = this::onProcessingTime;
Review Comment:
Is it feasible to make `onProcessingTime` package-private and override this in `InternalTimerServiceAsyncImpl `?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceAsyncImpl.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.asyncprocessing.AsyncExecutionController;
+import org.apache.flink.runtime.asyncprocessing.RecordContext;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import static org.apache.flink.runtime.asyncprocessing.KeyAccountingUnit.EMPTY_RECORD;
+
+/**
+ * An implementation of {@link InternalTimerService} that is used by {@link
+ * org.apache.flink.streaming.runtime.operators.asyncprocessing.AbstractAsyncStateStreamOperator}.
+ * The timer service will set {@link RecordContext} for the timers before invoking action to
+ * preserve the execution order between timer firing and records processing.
+ *
+ * @see <a
+ * href=https://cwiki.apache.org/confluence/display/FLINK/FLIP-425%3A+Asynchronous+Execution+Model#FLIP425:AsynchronousExecutionModel-Timers>FLIP-425
+ * timers section.</a>
+ * @param <K> Type of timer's key.
+ * @param <N> Type of the namespace to which timers are scoped.
+ */
+public class InternalTimerServiceAsyncImpl<K, N> extends InternalTimerServiceImpl<K, N> {
+
+ private AsyncExecutionController<K> asyncExecutionController;
+
+ InternalTimerServiceAsyncImpl(
+ TaskIOMetricGroup taskIOMetricGroup,
+ KeyGroupRange localKeyGroupRange,
+ KeyContext keyContext,
+ ProcessingTimeService processingTimeService,
+ KeyGroupedInternalPriorityQueue processingTimeTimersQueue,
+ KeyGroupedInternalPriorityQueue eventTimeTimersQueue,
+ StreamTaskCancellationContext cancellationContext,
+ AsyncExecutionController<K> asyncExecutionController) {
+ super(
+ taskIOMetricGroup,
+ localKeyGroupRange,
+ keyContext,
+ processingTimeService,
+ processingTimeTimersQueue,
+ eventTimeTimersQueue,
+ cancellationContext);
+ this.asyncExecutionController = asyncExecutionController;
+ this.processingTimeCallback = this::onProcessingTime;
+ }
+
+ private void onProcessingTime(long time) throws Exception {
+ // null out the timer in case the Triggerable calls registerProcessingTimeTimer()
+ // inside the callback.
+ nextTimer = null;
+
+ InternalTimer<K, N> timer;
+
+ while ((timer = processingTimeTimersQueue.peek()) != null
+ && timer.getTimestamp() <= time
+ && !cancellationContext.isCancelled()) {
+ RecordContext<K> recordCtx =
+ asyncExecutionController.buildContext(EMPTY_RECORD, timer.getKey());
+ recordCtx.retain();
+ asyncExecutionController.setCurrentContext(recordCtx);
+ keyContext.setCurrentKey(timer.getKey());
+ processingTimeTimersQueue.poll();
+ final InternalTimer<K, N> timerToTrigger = timer;
+ asyncExecutionController.syncPointRequestWithCallback(
+ () -> triggerTarget.onProcessingTime(timerToTrigger));
+ taskIOMetricGroup.getNumFiredTimers().inc();
+ recordCtx.release();
+ }
+
+ if (timer != null && nextTimer == null) {
+ nextTimer =
+ processingTimeService.registerTimer(
+ timer.getTimestamp(), this::onProcessingTime);
+ }
+ }
+
+ /**
+ * Advance one watermark, this will fire some event timers.
+ *
+ * @param time the time in watermark.
+ */
+ @Override
+ public void advanceWatermark(long time) throws Exception {
+ currentWatermark = time;
+
+ InternalTimer<K, N> timer;
+
+ while ((timer = eventTimeTimersQueue.peek()) != null
+ && timer.getTimestamp() <= time
+ && !cancellationContext.isCancelled()) {
+ RecordContext<K> recordCtx =
+ asyncExecutionController.buildContext(EMPTY_RECORD, timer.getKey());
+ recordCtx.retain();
+ asyncExecutionController.setCurrentContext(recordCtx);
+ keyContext.setCurrentKey(timer.getKey());
+ eventTimeTimersQueue.poll();
+ final InternalTimer<K, N> timerToTrigger = timer;
+ asyncExecutionController.syncPointRequestWithCallback(
+ () -> triggerTarget.onEventTime(timerToTrigger));
+ taskIOMetricGroup.getNumFiredTimers().inc();
+ recordCtx.release();
+ }
+ }
+
+ protected void foreachTimer(
Review Comment:
Actually, providing batch operation in async execution is dangerous. This should be carefully evaluated. We could discuss this later.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceAsyncImpl.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.asyncprocessing.AsyncExecutionController;
+import org.apache.flink.runtime.asyncprocessing.RecordContext;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import static org.apache.flink.runtime.asyncprocessing.KeyAccountingUnit.EMPTY_RECORD;
+
+/**
+ * An implementation of {@link InternalTimerService} that is used by {@link
+ * org.apache.flink.streaming.runtime.operators.asyncprocessing.AbstractAsyncStateStreamOperator}.
+ * The timer service will set {@link RecordContext} for the timers before invoking action to
+ * preserve the execution order between timer firing and records processing.
+ *
+ * @see <a
+ * href=https://cwiki.apache.org/confluence/display/FLINK/FLIP-425%3A+Asynchronous+Execution+Model#FLIP425:AsynchronousExecutionModel-Timers>FLIP-425
+ * timers section.</a>
+ * @param <K> Type of timer's key.
+ * @param <N> Type of the namespace to which timers are scoped.
+ */
+public class InternalTimerServiceAsyncImpl<K, N> extends InternalTimerServiceImpl<K, N> {
+
+ private AsyncExecutionController<K> asyncExecutionController;
+
+ InternalTimerServiceAsyncImpl(
+ TaskIOMetricGroup taskIOMetricGroup,
+ KeyGroupRange localKeyGroupRange,
+ KeyContext keyContext,
+ ProcessingTimeService processingTimeService,
+ KeyGroupedInternalPriorityQueue processingTimeTimersQueue,
+ KeyGroupedInternalPriorityQueue eventTimeTimersQueue,
+ StreamTaskCancellationContext cancellationContext,
+ AsyncExecutionController<K> asyncExecutionController) {
+ super(
+ taskIOMetricGroup,
+ localKeyGroupRange,
+ keyContext,
+ processingTimeService,
+ processingTimeTimersQueue,
+ eventTimeTimersQueue,
+ cancellationContext);
+ this.asyncExecutionController = asyncExecutionController;
+ this.processingTimeCallback = this::onProcessingTime;
+ }
+
+ private void onProcessingTime(long time) throws Exception {
+ // null out the timer in case the Triggerable calls registerProcessingTimeTimer()
+ // inside the callback.
+ nextTimer = null;
+
+ InternalTimer<K, N> timer;
+
+ while ((timer = processingTimeTimersQueue.peek()) != null
+ && timer.getTimestamp() <= time
+ && !cancellationContext.isCancelled()) {
+ RecordContext<K> recordCtx =
+ asyncExecutionController.buildContext(EMPTY_RECORD, timer.getKey());
+ recordCtx.retain();
+ asyncExecutionController.setCurrentContext(recordCtx);
+ keyContext.setCurrentKey(timer.getKey());
+ processingTimeTimersQueue.poll();
+ final InternalTimer<K, N> timerToTrigger = timer;
+ asyncExecutionController.syncPointRequestWithCallback(
+ () -> triggerTarget.onProcessingTime(timerToTrigger));
+ taskIOMetricGroup.getNumFiredTimers().inc();
+ recordCtx.release();
Review Comment:
Can we extract the common part of `onProcessingTime`, `advanceWatermark` and `foreachTimer` into one helper method? e.g. `maintainContextAndProcess(K key, ThrowingRunnable<Exception> runnable)`.
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24672:
URL: https://github.com/apache/flink/pull/24672#discussion_r1568659887
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManager.java:
##########
@@ -51,6 +52,21 @@ <N> InternalTimerService<N> getInternalTimerService(
TypeSerializer<N> namespaceSerializer,
Triggerable<K, N> triggerable);
+ /**
+ * Creates an {@link InternalTimerServiceAsyncImpl} for handling a group of timers identified by
+ * the given {@code name}. The timers are scoped to a key and namespace. Mainly used by async
+ * operators.
+ *
+ * <p>Some essential order preservation will be added when the given {@link Triggerable} is
+ * invoked.
+ */
+ <N> InternalTimerService<N> getAsyncInternalTimerService(
+ String name,
+ TypeSerializer<K> keySerializer,
+ TypeSerializer<N> namespaceSerializer,
+ Triggerable<K, N> triggerable,
+ AsyncExecutionController asyncExecutionController);
Review Comment:
It is better to provide the type parameter `K` from here and any other methods of this class.
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #24672:
URL: https://github.com/apache/flink/pull/24672#discussion_r1569902836
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java:
##########
@@ -117,6 +120,7 @@ public class InternalTimerServiceImpl<K, N> implements InternalTimerService<N> {
startIdx = Math.min(keyGroupIdx, startIdx);
}
this.localKeyGroupRangeStartIdx = startIdx;
+ this.processingTimeCallback = this::onProcessingTime;
Review Comment:
How about making `onProcessingTime` a protected method? This way we won't need to introduce processingTimeCallback.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManagerImpl.java:
##########
@@ -179,11 +180,60 @@ <N> InternalTimerServiceImpl<K, N> registerOrGetTimerService(
return timerService;
}
+ @Override
+ public <N> InternalTimerService<N> getAsyncInternalTimerService(
+ String name,
+ TypeSerializer<K> keySerializer,
+ TypeSerializer<N> namespaceSerializer,
+ Triggerable<K, N> triggerable,
+ AsyncExecutionController<K> asyncExecutionController) {
+ checkNotNull(keySerializer, "Timers can only be used on keyed operators.");
+
+ // the following casting is to overcome type restrictions.
+ TimerSerializer<K, N> timerSerializer =
+ new TimerSerializer<>(keySerializer, namespaceSerializer);
+
+ InternalTimerServiceAsyncImpl<K, N> timerService =
+ registerOrGetAsyncTimerService(name, timerSerializer, asyncExecutionController);
+
+ timerService.startTimerService(
+ timerSerializer.getKeySerializer(),
+ timerSerializer.getNamespaceSerializer(),
+ triggerable);
+
+ return timerService;
+ }
+
+ <N> InternalTimerServiceAsyncImpl<K, N> registerOrGetAsyncTimerService(
+ String name,
+ TimerSerializer<K, N> timerSerializer,
+ AsyncExecutionController<K> asyncExecutionController) {
+ InternalTimerServiceAsyncImpl<K, N> timerService =
+ (InternalTimerServiceAsyncImpl<K, N>) timerServices.get(name);
+ if (timerService == null) {
+
+ timerService =
+ new InternalTimerServiceAsyncImpl<>(
+ taskIOMetricGroup,
+ localKeyGroupRange,
+ keyContext,
+ processingTimeService,
+ createTimerPriorityQueue(
+ PROCESSING_TIMER_PREFIX + name, timerSerializer),
+ createTimerPriorityQueue(EVENT_TIMER_PREFIX + name, timerSerializer),
+ cancellationContext,
+ asyncExecutionController);
+
+ timerServices.put(name, timerService);
+ }
+ return timerService;
+ }
+
Map<String, InternalTimerServiceImpl<K, ?>> getRegisteredTimerServices() {
return Collections.unmodifiableMap(timerServices);
}
- private <N>
+ protected <N>
KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> createTimerPriorityQueue(
Review Comment:
This method seems not used in subclasses, so `private` might be enough. Same for `restoreStateForKeyGroup`.
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24672:
URL: https://github.com/apache/flink/pull/24672#discussion_r1571754617
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceAsyncImpl.java:
##########
@@ -0,0 +1,142 @@
+/*
+ * 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.asyncprocessing.AsyncExecutionController;
+import org.apache.flink.runtime.asyncprocessing.RecordContext;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.BiConsumerWithException;
+import org.apache.flink.util.function.ThrowingRunnable;
+
+/**
+ * An implementation of {@link InternalTimerService} that is used by {@link
+ * org.apache.flink.streaming.runtime.operators.asyncprocessing.AbstractAsyncStateStreamOperator}.
+ * The timer service will set {@link RecordContext} for the timers before invoking action to
+ * preserve the execution order between timer firing and records processing.
+ *
+ * @see <a
+ * href=https://cwiki.apache.org/confluence/display/FLINK/FLIP-425%3A+Asynchronous+Execution+Model#FLIP425:AsynchronousExecutionModel-Timers>FLIP-425
+ * timers section.</a>
+ * @param <K> Type of timer's key.
+ * @param <N> Type of the namespace to which timers are scoped.
+ */
+@Internal
+public class InternalTimerServiceAsyncImpl<K, N> extends InternalTimerServiceImpl<K, N> {
+
+ private AsyncExecutionController<K> asyncExecutionController;
+
+ InternalTimerServiceAsyncImpl(
+ TaskIOMetricGroup taskIOMetricGroup,
+ KeyGroupRange localKeyGroupRange,
+ KeyContext keyContext,
+ ProcessingTimeService processingTimeService,
+ KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> processingTimeTimersQueue,
+ KeyGroupedInternalPriorityQueue<TimerHeapInternalTimer<K, N>> eventTimeTimersQueue,
+ StreamTaskCancellationContext cancellationContext,
+ AsyncExecutionController<K> asyncExecutionController) {
+ super(
+ taskIOMetricGroup,
+ localKeyGroupRange,
+ keyContext,
+ processingTimeService,
+ processingTimeTimersQueue,
+ eventTimeTimersQueue,
+ cancellationContext);
+ this.asyncExecutionController = asyncExecutionController;
+ }
+
+ void onProcessingTime(long time) throws Exception {
Review Comment:
add `@Override` here?
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimeServiceManagerImpl.java:
##########
@@ -179,6 +180,55 @@ <N> InternalTimerServiceImpl<K, N> registerOrGetTimerService(
return timerService;
}
+ @Override
+ public <N> InternalTimerService<N> getAsyncInternalTimerService(
+ String name,
+ TypeSerializer<K> keySerializer,
+ TypeSerializer<N> namespaceSerializer,
+ Triggerable<K, N> triggerable,
+ AsyncExecutionController<K> asyncExecutionController) {
+ checkNotNull(keySerializer, "Timers can only be used on keyed operators.");
+
+ // the following casting is to overcome type restrictions.
+ TimerSerializer<K, N> timerSerializer =
+ new TimerSerializer<>(keySerializer, namespaceSerializer);
+
+ InternalTimerServiceAsyncImpl<K, N> timerService =
+ registerOrGetAsyncTimerService(name, timerSerializer, asyncExecutionController);
+
+ timerService.startTimerService(
+ timerSerializer.getKeySerializer(),
+ timerSerializer.getNamespaceSerializer(),
+ triggerable);
+
+ return timerService;
+ }
+
+ <N> InternalTimerServiceAsyncImpl<K, N> registerOrGetAsyncTimerService(
+ String name,
+ TimerSerializer<K, N> timerSerializer,
+ AsyncExecutionController<K> asyncExecutionController) {
+ InternalTimerServiceAsyncImpl<K, N> timerService =
+ (InternalTimerServiceAsyncImpl<K, N>) timerServices.get(name);
+ if (timerService == null) {
+
Review Comment:
nit. remove this empty line
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on PR #24672:
URL: https://github.com/apache/flink/pull/24672#issuecomment-2066225062
@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-35028][runtime] Timer firing under async execution model [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia merged PR #24672:
URL: https://github.com/apache/flink/pull/24672
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #24672:
URL: https://github.com/apache/flink/pull/24672#discussion_r1568257919
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceAsyncImpl.java:
##########
@@ -0,0 +1,145 @@
+/*
+ * 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.asyncprocessing.AsyncExecutionController;
+import org.apache.flink.runtime.asyncprocessing.RecordContext;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import static org.apache.flink.runtime.asyncprocessing.KeyAccountingUnit.EMPTY_RECORD;
+
+/**
+ * An implementation of {@link InternalTimerService} that is used by {@link
+ * org.apache.flink.streaming.runtime.operators.asyncprocessing.AbstractAsyncStateStreamOperator}.
+ * The timer service will set {@link RecordContext} for the timers before invoking action to
+ * preserve the execution order between timer firing and records processing.
+ *
+ * @see <a
+ * href=https://cwiki.apache.org/confluence/display/FLINK/FLIP-425%3A+Asynchronous+Execution+Model#FLIP425:AsynchronousExecutionModel-Timers>FLIP-425
+ * timers section.
+ * @param <K> Type of timer's key.
+ * @param <N> Type of the namespace to which timers are scoped.
+ */
+public class InternalTimerServiceAsyncImpl<K, N> extends InternalTimerServiceImpl<K, N> {
+
+ private AsyncExecutionController asyncExecutionController;
+
+ InternalTimerServiceAsyncImpl(
+ TaskIOMetricGroup taskIOMetricGroup,
+ KeyGroupRange localKeyGroupRange,
+ KeyContext keyContext,
+ ProcessingTimeService processingTimeService,
+ KeyGroupedInternalPriorityQueue processingTimeTimersQueue,
+ KeyGroupedInternalPriorityQueue eventTimeTimersQueue,
+ StreamTaskCancellationContext cancellationContext,
+ AsyncExecutionController asyncExecutionController) {
+ super(
+ taskIOMetricGroup,
+ localKeyGroupRange,
+ keyContext,
+ processingTimeService,
+ processingTimeTimersQueue,
+ eventTimeTimersQueue,
+ cancellationContext);
+ this.asyncExecutionController = asyncExecutionController;
+ this.processingTimeCallback = this::onAsyncProcessingTime;
+ }
+
+ private void onAsyncProcessingTime(long time) throws Exception {
+ // null out the timer in case the Triggerable calls registerProcessingTimeTimer()
+ // inside the callback.
+ nextTimer = null;
+
+ InternalTimer<K, N> timer;
+
+ while ((timer = processingTimeTimersQueue.peek()) != null
+ && timer.getTimestamp() <= time
+ && !cancellationContext.isCancelled()) {
+ RecordContext<K> recordCtx =
+ asyncExecutionController.buildContext(EMPTY_RECORD, timer.getKey());
+ recordCtx.retain();
+ asyncExecutionController.setCurrentContext(recordCtx);
+ keyContext.setCurrentKey(timer.getKey());
+ processingTimeTimersQueue.poll();
+ final InternalTimer<K, N> timerToTrigger = timer;
+ asyncExecutionController.syncPointRequestWithCallback(
+ () -> triggerTarget.onProcessingTime(timerToTrigger));
+ taskIOMetricGroup.getNumFiredTimers().inc();
+ recordCtx.release();
+ }
+
+ if (timer != null && nextTimer == null) {
+ nextTimer =
+ processingTimeService.registerTimer(
+ timer.getTimestamp(), this::onAsyncProcessingTime);
+ }
+ }
+
+ /**
+ * Advance one watermark, this will fire some event timers.
+ *
+ * @param time the time in watermark.
+ */
+ public void advanceWatermark(long time) throws Exception {
Review Comment:
nit: it might be better to add `@Override` to this and the following method.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/runtime/operators/asyncprocessing/AbstractAsyncStateStreamOperator.java:
##########
@@ -133,6 +140,40 @@ public final <T> ThrowingConsumer<StreamRecord<T>, Exception> getRecordProcessor
getClass().getName(), inputId));
}
+ /**
+ * Returns a {@link org.apache.flink.streaming.api.operators.InternalTimerServiceAsyncImpl} that
+ * can be used to query current processing time and event time and to set timers. An operator
+ * can have several timer services, where each has its own namespace serializer.
+ *
+ * <p>Timers are always scoped to a key, the currently active key of a keyed stream operation.
+ * When a timer fires, this key will also be set as the currently active key, and a {@link
+ * RecordContext} will also be set as the current processing context to preserve record order.
Review Comment:
It might be better to keep descriptions to implementation details as inline comments instead of as JavaDocs.
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java:
##########
@@ -117,6 +120,7 @@ public class InternalTimerServiceImpl<K, N> implements InternalTimerService<N> {
startIdx = Math.min(keyGroupIdx, startIdx);
}
this.localKeyGroupRangeStartIdx = startIdx;
+ this.processingTimeCallback = this::onProcessingTime;
Review Comment:
It might be better to avoid introducing `processingTimeCallback`, or introduce it and make it a final variable. Given that the PR has already reused the name `advanceWatermark` in `InternalTimerServiceAsyncImpl` instead of using `asyncAdvanceWatermark`, it might be better to rename `onAsyncProcessingTime` back to `onProcessingTime` as well.
##########
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/InternalTimerServiceAsyncImplTest.java:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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.TypeSerializer;
+import org.apache.flink.api.common.typeutils.base.IntSerializer;
+import org.apache.flink.api.common.typeutils.base.StringSerializer;
+import org.apache.flink.runtime.asyncprocessing.AsyncExecutionController;
+import org.apache.flink.runtime.asyncprocessing.RecordContext;
+import org.apache.flink.runtime.asyncprocessing.StateExecutor;
+import org.apache.flink.runtime.asyncprocessing.StateRequest;
+import org.apache.flink.runtime.asyncprocessing.StateRequestType;
+import org.apache.flink.runtime.mailbox.SyncMailboxExecutor;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.runtime.metrics.groups.UnregisteredMetricGroups;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.PriorityQueueSetFactory;
+import org.apache.flink.runtime.state.heap.HeapPriorityQueueSetFactory;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.streaming.runtime.tasks.TestProcessingTimeService;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.concurrent.CompletableFuture;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link InternalTimerServiceAsyncImpl}. */
+class InternalTimerServiceAsyncImplTest {
Review Comment:
It might be better to make this class extend from `InternalTimerServiceImplTest` to reuse the existing test cases.
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #24672:
URL: https://github.com/apache/flink/pull/24672#discussion_r1568731699
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceImpl.java:
##########
@@ -117,6 +120,7 @@ public class InternalTimerServiceImpl<K, N> implements InternalTimerService<N> {
startIdx = Math.min(keyGroupIdx, startIdx);
}
this.localKeyGroupRangeStartIdx = startIdx;
+ this.processingTimeCallback = this::onProcessingTime;
Review Comment:
The original idea was to avoid rewriting `startTimerService()` and `registerProcessingTimeTimer` by introducing `processingTimeCallback`, since it needs to be reassigned in the subclass, it cannot be marked as `final`.
For the `onAsyncProcessingTime`, I changed it back to `onProcessingTime`, BTW, `onProcessingTime()` is a private method.
--
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-35028][runtime] Timer firing under async execution model [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #24672:
URL: https://github.com/apache/flink/pull/24672#discussion_r1570140913
##########
flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/InternalTimerServiceAsyncImpl.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * 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.asyncprocessing.AsyncExecutionController;
+import org.apache.flink.runtime.asyncprocessing.RecordContext;
+import org.apache.flink.runtime.metrics.groups.TaskIOMetricGroup;
+import org.apache.flink.runtime.state.KeyGroupRange;
+import org.apache.flink.runtime.state.KeyGroupedInternalPriorityQueue;
+import org.apache.flink.streaming.runtime.tasks.ProcessingTimeService;
+import org.apache.flink.streaming.runtime.tasks.StreamTaskCancellationContext;
+import org.apache.flink.util.CloseableIterator;
+import org.apache.flink.util.function.BiConsumerWithException;
+
+import static org.apache.flink.runtime.asyncprocessing.KeyAccountingUnit.EMPTY_RECORD;
+
+/**
+ * An implementation of {@link InternalTimerService} that is used by {@link
+ * org.apache.flink.streaming.runtime.operators.asyncprocessing.AbstractAsyncStateStreamOperator}.
+ * The timer service will set {@link RecordContext} for the timers before invoking action to
+ * preserve the execution order between timer firing and records processing.
+ *
+ * @see <a
+ * href=https://cwiki.apache.org/confluence/display/FLINK/FLIP-425%3A+Asynchronous+Execution+Model#FLIP425:AsynchronousExecutionModel-Timers>FLIP-425
+ * timers section.</a>
+ * @param <K> Type of timer's key.
+ * @param <N> Type of the namespace to which timers are scoped.
+ */
+public class InternalTimerServiceAsyncImpl<K, N> extends InternalTimerServiceImpl<K, N> {
+
+ private AsyncExecutionController<K> asyncExecutionController;
+
+ InternalTimerServiceAsyncImpl(
+ TaskIOMetricGroup taskIOMetricGroup,
+ KeyGroupRange localKeyGroupRange,
+ KeyContext keyContext,
+ ProcessingTimeService processingTimeService,
+ KeyGroupedInternalPriorityQueue processingTimeTimersQueue,
+ KeyGroupedInternalPriorityQueue eventTimeTimersQueue,
+ StreamTaskCancellationContext cancellationContext,
+ AsyncExecutionController<K> asyncExecutionController) {
+ super(
+ taskIOMetricGroup,
+ localKeyGroupRange,
+ keyContext,
+ processingTimeService,
+ processingTimeTimersQueue,
+ eventTimeTimersQueue,
+ cancellationContext);
+ this.asyncExecutionController = asyncExecutionController;
+ this.processingTimeCallback = this::onProcessingTime;
+ }
+
+ private void onProcessingTime(long time) throws Exception {
+ // null out the timer in case the Triggerable calls registerProcessingTimeTimer()
+ // inside the callback.
+ nextTimer = null;
+
+ InternalTimer<K, N> timer;
+
+ while ((timer = processingTimeTimersQueue.peek()) != null
+ && timer.getTimestamp() <= time
+ && !cancellationContext.isCancelled()) {
+ RecordContext<K> recordCtx =
+ asyncExecutionController.buildContext(EMPTY_RECORD, timer.getKey());
+ recordCtx.retain();
+ asyncExecutionController.setCurrentContext(recordCtx);
+ keyContext.setCurrentKey(timer.getKey());
+ processingTimeTimersQueue.poll();
+ final InternalTimer<K, N> timerToTrigger = timer;
+ asyncExecutionController.syncPointRequestWithCallback(
+ () -> triggerTarget.onProcessingTime(timerToTrigger));
+ taskIOMetricGroup.getNumFiredTimers().inc();
+ recordCtx.release();
+ }
+
+ if (timer != null && nextTimer == null) {
+ nextTimer =
+ processingTimeService.registerTimer(
+ timer.getTimestamp(), this::onProcessingTime);
+ }
+ }
+
+ /**
+ * Advance one watermark, this will fire some event timers.
+ *
+ * @param time the time in watermark.
+ */
+ @Override
+ public void advanceWatermark(long time) throws Exception {
+ currentWatermark = time;
+
+ InternalTimer<K, N> timer;
+
+ while ((timer = eventTimeTimersQueue.peek()) != null
+ && timer.getTimestamp() <= time
+ && !cancellationContext.isCancelled()) {
+ RecordContext<K> recordCtx =
+ asyncExecutionController.buildContext(EMPTY_RECORD, timer.getKey());
+ recordCtx.retain();
+ asyncExecutionController.setCurrentContext(recordCtx);
+ keyContext.setCurrentKey(timer.getKey());
+ eventTimeTimersQueue.poll();
+ final InternalTimer<K, N> timerToTrigger = timer;
+ asyncExecutionController.syncPointRequestWithCallback(
+ () -> triggerTarget.onEventTime(timerToTrigger));
+ taskIOMetricGroup.getNumFiredTimers().inc();
+ recordCtx.release();
+ }
+ }
+
+ protected void foreachTimer(
Review Comment:
`foreachTimer` is mainly used for `state-processor-api` to read timers from savepoint/checkpoint, maybe we can restrict state-processor-api to only use sync state API in later. I add a TODO here.
--
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