You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Zakelly (via GitHub)" <gi...@apache.org> on 2024/04/03 10:45:39 UTC
[PR] F34986 [flink]
Zakelly opened a new pull request, #24614:
URL: https://github.com/apache/flink/pull/24614
## What is the purpose of the change
This PR ship the core part of FLIP-425, including the basic execution logic of AsyncExecutionController.
Note: This PR is based on #24597 which is still under review.
## Brief change log
- AsyncExecutionController and other components around it.
- RecordContext and reference counting mechanism.
- Basic implementation of KeyAccountingUnit.
## Verifying this change
- Added unit tests under `org.apache.flink.runtime.asyncprocessing`
## 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): yes (introduce new)
- 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? 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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1554814874
##########
flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/ContextStateFutureImplTest.java:
##########
@@ -0,0 +1,234 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.core.state.StateFutureUtils;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.LinkedList;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link ContextStateFutureImpl}. */
+public class ContextStateFutureImplTest {
+
+ @Test
+ public void testThenApply() {
+ SingleStepRunner runner = new SingleStepRunner();
+ KeyAccountingUnit<String, String> keyAccountingUnit = new KeyAccountingUnit<>();
+ RecordContext<String, String> recordContext =
+ new RecordContext<>(keyAccountingUnit, "a", "b");
+
+ // validate
+ ContextStateFutureImpl<Void> future =
+ new ContextStateFutureImpl<>(runner::submit, recordContext);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(1);
+ future.thenApply((v) -> 1L);
+ future.complete(null);
+ assertThat(runner.runThrough()).isTrue();
+ assertThat(recordContext.getReferenceCount()).isEqualTo(0);
+
+ // validate completion before callback
+ future = new ContextStateFutureImpl<>(runner::submit, recordContext);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(1);
+ future.complete(null);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(0);
+ future.thenApply((v) -> 1L);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(0);
+ assertThat(runner.runThrough()).isFalse();
+ }
+
+ @Test
+ public void testThenAccept() {
+ SingleStepRunner runner = new SingleStepRunner();
+ KeyAccountingUnit<String, String> keyAccountingUnit = new KeyAccountingUnit<>();
+ RecordContext<String, String> recordContext =
+ new RecordContext<>(keyAccountingUnit, "a", "b");
+
+ // validate
+ ContextStateFutureImpl<Void> future =
+ new ContextStateFutureImpl<>(runner::submit, recordContext);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(1);
+ future.thenAccept((v) -> {});
+ future.complete(null);
+ assertThat(runner.runThrough()).isTrue();
+ assertThat(recordContext.getReferenceCount()).isEqualTo(0);
+
+ // validate completion before callback
+ future = new ContextStateFutureImpl<>(runner::submit, recordContext);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(1);
+ future.complete(null);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(0);
+ future.thenAccept((v) -> {});
+ assertThat(recordContext.getReferenceCount()).isEqualTo(0);
+ assertThat(runner.runThrough()).isFalse();
+ }
+
+ @Test
+ public void testThenCompose() {
+ SingleStepRunner runner = new SingleStepRunner();
+ KeyAccountingUnit<String, String> keyAccountingUnit = new KeyAccountingUnit<>();
+ RecordContext<String, String> recordContext =
+ new RecordContext<>(keyAccountingUnit, "a", "b");
+
+ // validate
+ ContextStateFutureImpl<Void> future =
+ new ContextStateFutureImpl<>(runner::submit, recordContext);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(1);
+ future.thenCompose((v) -> StateFutureUtils.completedFuture(1L));
+ future.complete(null);
+ assertThat(runner.runThrough()).isTrue();
+ assertThat(recordContext.getReferenceCount()).isEqualTo(0);
+
+ // validate completion before callback
+ future = new ContextStateFutureImpl<>(runner::submit, recordContext);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(1);
+ future.complete(null);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(0);
+ future.thenCompose((v) -> StateFutureUtils.completedFuture(1L));
+ assertThat(recordContext.getReferenceCount()).isEqualTo(0);
+ assertThat(runner.runThrough()).isFalse();
+ }
+
+ @Test
+ public void testThenCombine() {
+ SingleStepRunner runner = new SingleStepRunner();
+ KeyAccountingUnit<String, String> keyAccountingUnit = new KeyAccountingUnit<>();
+ RecordContext<String, String> recordContext =
+ new RecordContext<>(keyAccountingUnit, "a", "b");
+
+ // validate
+ ContextStateFutureImpl<Void> future1 =
+ new ContextStateFutureImpl<>(runner::submit, recordContext);
+ ContextStateFutureImpl<Void> future2 =
+ new ContextStateFutureImpl<>(runner::submit, recordContext);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(2);
+ future1.thenCombine(future2, (v1, v2) -> 1L);
+ future1.complete(null);
+ future2.complete(null);
+ assertThat(runner.runThrough()).isTrue();
+ assertThat(recordContext.getReferenceCount()).isEqualTo(0);
+
+ // validate future1 completion before callback
+ future1 = new ContextStateFutureImpl<>(runner::submit, recordContext);
+ future2 = new ContextStateFutureImpl<>(runner::submit, recordContext);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(2);
+ future1.complete(null);
+ future1.thenCombine(future2, (v1, v2) -> 1L);
+ assertThat(recordContext.getReferenceCount()).isGreaterThan(1);
+ future2.complete(null);
+ assertThat(runner.runThrough()).isTrue();
+ assertThat(recordContext.getReferenceCount()).isEqualTo(0);
+
+ // validate future2 completion before callback
+ future1 = new ContextStateFutureImpl<>(runner::submit, recordContext);
+ future2 = new ContextStateFutureImpl<>(runner::submit, recordContext);
+ assertThat(recordContext.getReferenceCount()).isEqualTo(2);
+ future2.complete(null);
+ future1.thenCombine(future2, (v1, v2) -> 1L);
+ assertThat(recordContext.getReferenceCount()).isGreaterThan(1);
+ future1.complete(null);
+ assertThat(runner.runThrough()).isTrue();
+ assertThat(recordContext.getReferenceCount()).isEqualTo(0);
+ }
+
+ @Test
+ public void testComplex() {
+ SingleStepRunner runner = new SingleStepRunner();
+ KeyAccountingUnit<String, String> keyAccountingUnit = new KeyAccountingUnit<>();
+ RecordContext<String, String> recordContext =
+ new RecordContext<>(keyAccountingUnit, "a", "b");
+
+ for (int i = 0; i < 32; i++) { // 2^5 for completion status combination
Review Comment:
Add some descriptions to explain what the bits in `i` represent?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java:
##########
@@ -0,0 +1,292 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.StateFuture;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.core.state.StateFutureUtils;
+import org.apache.flink.runtime.asyncprocessing.StateRequest.RequestType;
+import org.apache.flink.runtime.mailbox.SyncMailboxExecutor;
+import org.apache.flink.util.Preconditions;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+/** Test for {@link AsyncExecutionController}. */
+class AsyncExecutionControllerTest {
+
+ // TODO: this test is not well completed, cause buffering in AEC is not implemented.
+ // Yet, just for illustrating the interaction between AEC and Async state API.
+ @Test
+ void testBasicRun() {
+ TestAsyncExecutionController<String, String> aec =
+ new TestAsyncExecutionController<>(
+ new SyncMailboxExecutor(), new TestStateExecutor());
+ TestUnderlyingState underlyingState = new TestUnderlyingState();
+ TestValueState valueState = new TestValueState(aec, underlyingState);
+ AtomicInteger output = new AtomicInteger();
+ Runnable userCode =
+ () -> {
+ valueState
+ .asyncValue()
+ .thenCompose(
+ val -> {
+ int updated = (val == null ? 1 : (val + 1));
+ return valueState
+ .asyncUpdate(updated)
+ .thenCompose(
+ o ->
+ StateFutureUtils.completedFuture(
+ updated));
+ })
+ .thenAccept(val -> output.set(val));
+ };
+
+ // ============================ element1 ============================
+ String record1 = "key1-r1";
+ String key1 = "key1";
+ // Simulate the wrapping in {@link RecordProcessorUtils#getRecordProcessor()}, wrapping the
+ // record and key with RecordContext.
+ RecordContext<String, String> recordContext1 = aec.buildContext(record1, key1);
+ aec.setCurrentContext(recordContext1);
+ // execute user code
+ userCode.run();
+
+ // single-step run
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
Review Comment:
Add some descriptions here to explain what the contents of each step in active buffer are?
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1556830946
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.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.runtime.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
+
+ /** The corresponding context that currently runs in task thread. */
+ RecordContext<R, K> currentContext;
+
+ public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) {
+ this(mailboxExecutor, stateExecutor, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM);
+ }
+
+ public AsyncExecutionController(
+ MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int maxInFlightRecords) {
+ this.keyAccountingUnit = new KeyAccountingUnit<>();
+ this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor);
+ this.stateExecutor = stateExecutor;
+ this.maxInFlightRecordNum = maxInFlightRecords;
+ LOG.info("Create AsyncExecutionController: maxInFlightRecordsNum {}", maxInFlightRecords);
+ }
+
+ /**
+ * Build a new context based on record and key. Also wired with internal {@link
+ * KeyAccountingUnit}.
+ *
+ * @param record the given record.
+ * @param key the given key.
+ * @return the built record context.
+ */
+ public RecordContext<R, K> buildContext(R record, K key) {
Review Comment:
Thanks for the explanation.
I am just thinking about whether the `RecordContext` will be used by other classes or it just be maintained and passed internally. Maybe it could just be `generateContext` in the AEC.
Of course, I think we could remain this currently.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #24614:
URL: https://github.com/apache/flink/pull/24614#issuecomment-2041280097
Rebased and pushed 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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1557595222
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
+
+ /** The corresponding context that currently runs in task thread. */
+ RecordContext<R, K> currentContext;
Review Comment:
You're right about "fully managed", but I'm saying it is fully managed by `AEC` AND **related runtime components**. That means this is managed by processing framework instead of the caller of `handleRequest`, which is the `State`-related implementation.
The `buildContext` will be invoked right before the `processElement`. To be more specific, my brief idea is, both `buildContext` and `processElement` will be packed into one lambda in `RecordProcessorUtils#getRecordProcessor`.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #24614:
URL: https://github.com/apache/flink/pull/24614#issuecomment-2041280495
@fredia @masteryhx would you please take a 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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1555247481
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
+
+ /** The type of processing request. */
+ public enum RequestType {
+ /** Process one record without state access. */
+ SYNC,
+ /** Get from one {@link State}. */
+ GET,
+ /** Put to one {@link State}. */
+ PUT,
+ /** Merge value to an exist key in {@link State}. Mainly used for listState. */
+ MERGE,
+ /** Delete from one {@link State}. */
+ DELETE
+ }
+
+ /** The underlying state to be accessed, can be empty for {@link RequestType#SYNC}. */
+ @Nullable private final State state;
+
+ /** The type of this request. */
+ private final RequestType type;
+
+ /** The payload(input) of this request. */
+ @Nullable private final IN payload;
+
+ /** The future to collect the result of the request. */
+ private InternalStateFuture<OUT> stateFuture;
+
+ /** The record context of this request. */
+ private RecordContext<?, K> context;
+
+ StateRequest(@Nullable State state, RequestType type, @Nullable IN payload) {
+ this.state = state;
+ this.type = type;
+ this.payload = payload;
+ }
+
+ RequestType getRequestType() {
+ return type;
+ }
+
+ @Nullable
+ IN getPayload() {
+ return payload;
+ }
+
+ @Nullable
+ State getState() {
+ return state;
+ }
+
+ InternalStateFuture<OUT> getFuture() {
+ return stateFuture;
+ }
+
+ void setFuture(InternalStateFuture<OUT> future) {
+ stateFuture = future;
+ }
+
+ RecordContext<?, K> getRecordContext() {
+ return context;
+ }
+
+ void setRecordContext(RecordContext<?, K> context) {
+ this.context = context;
+ }
Review Comment:
Well this is a good suggestion. I'll try this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1555340649
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
Review Comment:
Fine, Considering it after introduing internal KV interface also makes sense.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1555175989
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
+
+ /** The type of processing request. */
+ public enum RequestType {
+ /** Process one record without state access. */
+ SYNC,
Review Comment:
The naming is a bit confusing.
I may thinked it as accessing state synchronously if I missed above comment.
How about `NOP` or other names at least says `without state access`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateExecutor.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.annotation.Internal;
+
+import java.util.concurrent.CompletableFuture;
+
+/** Executor for executing batch {@link StateRequest}s. */
+@Internal
+public interface StateExecutor {
+ /**
+ * Execute a batch of state requests.
+ *
+ * @param processingRequests the given batch of processing requests
+ * @return A future can determine whether execution has completed.
+ */
+ CompletableFuture<Boolean> executeBatchRequests(
+ Iterable<StateRequest<?, ?, ?>> processingRequests);
Review Comment:
Different state types could be executed together, right ?
How could we get specific state operations ? Current `RequestType` seems not enough.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.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.runtime.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
+
+ /** The corresponding context that currently runs in task thread. */
+ RecordContext<R, K> currentContext;
+
+ public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) {
+ this(mailboxExecutor, stateExecutor, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM);
+ }
+
+ public AsyncExecutionController(
+ MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int maxInFlightRecords) {
+ this.keyAccountingUnit = new KeyAccountingUnit<>();
+ this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor);
+ this.stateExecutor = stateExecutor;
+ this.maxInFlightRecordNum = maxInFlightRecords;
+ LOG.info("Create AsyncExecutionController: maxInFlightRecordsNum {}", maxInFlightRecords);
+ }
+
+ /**
+ * Build a new context based on record and key. Also wired with internal {@link
+ * KeyAccountingUnit}.
+ *
+ * @param record the given record.
+ * @param key the given key.
+ * @return the built record context.
+ */
+ public RecordContext<R, K> buildContext(R record, K key) {
Review Comment:
Could we unify this method with `setCurrentContext` ?
Of course we could provide `getCurrentContext` only for tests.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
Review Comment:
How about also adding type for `State`, e.g. `StateRequest<S extends State, K, IN, OUT>` ?
Then we could also avoid some explicit casting when `getState`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
+
+ /** The type of processing request. */
+ public enum RequestType {
+ /** Process one record without state access. */
+ SYNC,
+ /** Get from one {@link State}. */
+ GET,
+ /** Put to one {@link State}. */
+ PUT,
+ /** Merge value to an exist key in {@link State}. Mainly used for listState. */
+ MERGE,
+ /** Delete from one {@link State}. */
+ DELETE
+ }
+
+ /** The underlying state to be accessed, can be empty for {@link RequestType#SYNC}. */
+ @Nullable private final State state;
+
+ /** The type of this request. */
+ private final RequestType type;
+
+ /** The payload(input) of this request. */
+ @Nullable private final IN payload;
+
+ /** The future to collect the result of the request. */
+ private InternalStateFuture<OUT> stateFuture;
+
+ /** The record context of this request. */
+ private RecordContext<?, K> context;
+
+ StateRequest(@Nullable State state, RequestType type, @Nullable IN payload) {
+ this.state = state;
+ this.type = type;
+ this.payload = payload;
+ }
+
+ RequestType getRequestType() {
+ return type;
+ }
+
+ @Nullable
+ IN getPayload() {
+ return payload;
+ }
+
+ @Nullable
+ State getState() {
+ return state;
+ }
+
+ InternalStateFuture<OUT> getFuture() {
+ return stateFuture;
+ }
+
+ void setFuture(InternalStateFuture<OUT> future) {
+ stateFuture = future;
+ }
+
+ RecordContext<?, K> getRecordContext() {
+ return context;
+ }
+
+ void setRecordContext(RecordContext<?, K> context) {
+ this.context = context;
+ }
Review Comment:
I am also thinking about whether we could avoid 'two-step setting'.
How about extracting a new class e.g. `StateRequestContext` for AEC?
Then State layer could only see `StateRequest`, AEC could also only set its Context.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/KeyAccountingUnit.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.runtime.asyncprocessing;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Key accounting unit holds the current in-flight key and tracks the corresponding ongoing records,
+ * which is used to preserve the ordering of independent chained {@link
+ * org.apache.flink.api.common.state.v2.StateFuture}.
+ *
+ * @param <R> the type of record
+ * @param <K> the type of key
+ */
+public class KeyAccountingUnit<R, K> {
Review Comment:
We may have other implementations or it may be a interface, right ?
Just a minor suggestion, How about adding TODO here ?
##########
flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java:
##########
@@ -0,0 +1,305 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.StateFuture;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.core.state.StateFutureUtils;
+import org.apache.flink.runtime.asyncprocessing.StateRequest.RequestType;
+import org.apache.flink.runtime.mailbox.SyncMailboxExecutor;
+import org.apache.flink.util.Preconditions;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+/** Test for {@link AsyncExecutionController}. */
+class AsyncExecutionControllerTest {
+
+ // TODO: this test is not well completed, cause buffering in AEC is not implemented.
+ // Yet, just for illustrating the interaction between AEC and Async state API.
+ @Test
+ void testBasicRun() {
+ TestAsyncExecutionController<String, String> aec =
+ new TestAsyncExecutionController<>(
+ new SyncMailboxExecutor(), new TestStateExecutor());
+ TestUnderlyingState underlyingState = new TestUnderlyingState();
+ TestValueState valueState = new TestValueState(aec, underlyingState);
+ AtomicInteger output = new AtomicInteger();
+ Runnable userCode =
+ () -> {
+ valueState
+ .asyncValue()
+ .thenCompose(
+ val -> {
+ int updated = (val == null ? 1 : (val + 1));
+ return valueState
+ .asyncUpdate(updated)
+ .thenCompose(
+ o ->
+ StateFutureUtils.completedFuture(
+ updated));
+ })
+ .thenAccept(val -> output.set(val));
+ };
+
+ // ============================ element1 ============================
+ String record1 = "key1-r1";
+ String key1 = "key1";
+ // Simulate the wrapping in {@link RecordProcessorUtils#getRecordProcessor()}, wrapping the
+ // record and key with RecordContext.
+ RecordContext<String, String> recordContext1 = aec.buildContext(record1, key1);
+ aec.setCurrentContext(recordContext1);
+ // execute user code
+ userCode.run();
+
+ // Single-step run.
+ // Firstly, the user code generates value get in active buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // After running, the value update is in active buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // Value update finishes.
+ assertThat(aec.activeBuffer.size()).isEqualTo(0);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(0);
+ assertThat(output.get()).isEqualTo(1);
+ assertThat(recordContext1.getReferenceCount()).isEqualTo(0);
+
+ // ============================ element 2 & 3 ============================
+ String record2 = "key1-r2";
+ String key2 = "key1";
+ RecordContext<String, String> recordContext2 = aec.buildContext(record2, key2);
+ aec.setCurrentContext(recordContext2);
+ // execute user code
+ userCode.run();
+
+ String record3 = "key1-r3";
+ String key3 = "key1";
+ RecordContext<String, String> recordContext3 = aec.buildContext(record3, key3);
+ aec.setCurrentContext(recordContext3);
+ // execute user code
+ userCode.run();
+
+ // Single-step run.
+ // Firstly, the user code for record2 generates value get in active buffer,
+ // while user code for record3 generates value get in blocking buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.blockingBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // After running, the value update for record2 is in active buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.blockingBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // Value update for record2 finishes. The value get for record3 is still in blocking status.
+ assertThat(aec.activeBuffer.size()).isEqualTo(0);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(0);
+ assertThat(output.get()).isEqualTo(2);
+ assertThat(recordContext2.getReferenceCount()).isEqualTo(0);
+ assertThat(aec.blockingBuffer.size()).isEqualTo(1);
+
+ aec.migrateBlockingToActive();
+ // Value get for record3 is ready for run.
+
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.blockingBuffer.size()).isEqualTo(0);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // After running, the value update for record3 is in active buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // Value update for record3 finishes.
+ assertThat(aec.activeBuffer.size()).isEqualTo(0);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(0);
+ assertThat(output.get()).isEqualTo(3);
+ assertThat(recordContext3.getReferenceCount()).isEqualTo(0);
+
+ // ============================ element4 ============================
+ String record4 = "key3-r3";
+ String key4 = "key3";
+ RecordContext<String, String> recordContext4 = aec.buildContext(record4, key4);
+ aec.setCurrentContext(recordContext4);
+ // execute user code
+ userCode.run();
+
+ // Single-step run for another key.
+ // Firstly, the user code generates value get in active buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // After running, the value update is in active buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // Value update finishes.
+ assertThat(aec.activeBuffer.size()).isEqualTo(0);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(0);
+ assertThat(output.get()).isEqualTo(1);
+ assertThat(recordContext4.getReferenceCount()).isEqualTo(0);
+ }
+
+ /**
+ * An AsyncExecutionController for testing purpose, which integrates with basic buffer
+ * mechanism.
+ */
+ static class TestAsyncExecutionController<R, K> extends AsyncExecutionController<R, K> {
+
+ LinkedList<StateRequest<K, ?, ?>> activeBuffer;
+
+ LinkedList<StateRequest<K, ?, ?>> blockingBuffer;
+
+ public TestAsyncExecutionController(
+ MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) {
+ super(mailboxExecutor, stateExecutor);
+ activeBuffer = new LinkedList<>();
+ blockingBuffer = new LinkedList<>();
+ }
+
+ @Override
+ <IN, OUT> void insertActiveBuffer(StateRequest<K, IN, OUT> request) {
+ activeBuffer.push(request);
+ }
+
+ <IN, OUT> void insertBlockingBuffer(StateRequest<K, IN, OUT> request) {
+ blockingBuffer.push(request);
+ }
+
+ void triggerIfNeeded(boolean force) {
+ if (!force) {
+ // Disable normal trigger, to perform single-step debugging and check.
+ return;
+ }
+ LinkedList<StateRequest<?, ?, ?>> toRun = new LinkedList<>(activeBuffer);
+ activeBuffer.clear();
+ stateExecutor.executeBatchRequests(toRun);
+ }
+
+ void migrateBlockingToActive() {
+ Iterator<StateRequest<K, ?, ?>> blockingIter = blockingBuffer.iterator();
+ while (blockingIter.hasNext()) {
+ StateRequest<K, ?, ?> request = blockingIter.next();
+ if (request.getRecordContext().tryOccupyKey()) {
+ insertActiveBuffer(request);
+ blockingIter.remove();
+ }
+ }
+ }
+ }
+
+ /** Simulate the underlying state that is actually used to execute the request. */
+ static class TestUnderlyingState {
+
+ private HashMap<String, Integer> hashMap;
+
+ public TestUnderlyingState() {
+ this.hashMap = new HashMap<>();
+ }
+
+ public Integer get(String key) {
+ return hashMap.get(key);
+ }
+
+ public void update(String key, Integer val) {
+ hashMap.put(key, val);
+ }
+ }
+
+ static class TestValueState implements ValueState<Integer> {
+
+ private AsyncExecutionController<String, String> asyncExecutionController;
+
+ private TestUnderlyingState underlyingState;
+
+ public TestValueState(
+ AsyncExecutionController<String, String> aec, TestUnderlyingState underlyingState) {
+ this.asyncExecutionController = aec;
+ this.underlyingState = underlyingState;
+ }
+
+ @Override
+ public StateFuture<Void> asyncClear() {
+ StateRequest<String, Void, Void> request =
Review Comment:
Generate `StateRequest` in State layer looks good to me.
But how could we handle `SYNC` type which is invisible for `State` ?
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #24614:
URL: https://github.com/apache/flink/pull/24614#issuecomment-2047789828
@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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1554811623
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
+
+ /** The type of processing request. */
+ public enum RequestType {
+ /** Process one record without state access. */
+ SYNC,
+ /** Get from one {@link State}. */
+ GET,
+ /** Put to one {@link State}. */
+ PUT,
+ /** Merge value to an exist key in {@link State}. Mainly used for listState. */
+ MERGE,
+ /** Delete from one {@link State}. */
+ DELETE
+ }
+
+ /** The underlying state to be accessed, can be empty for {@link RequestType#SYNC}. */
+ @Nullable private final State state;
Review Comment:
The reason we keep it `org.apache.flink.api.common.state.v2.State` here is we hope the direct implementation of KV (aka `internal key-value` layer) to be an unified light-weight one, which is only a simple bridge between user-interface to `AEC` (see `AsyncExecutionControllerTest#TestValueState` as an example). I think we could make it internal kv interface or class parameter in future but for now I suggest we keep it simple.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1555208732
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
Review Comment:
I think this question is somewhat the same as https://github.com/apache/flink/pull/24614#discussion_r1554806613.
In the future, an `internal` `key-value` layer will be introduced, we could make it internal kv interface or class parameter 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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #24614:
URL: https://github.com/apache/flink/pull/24614#issuecomment-2044191108
Rebased master. Extracting `StateRequestType` with more detailed enum entry definition.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #24614:
URL: https://github.com/apache/flink/pull/24614#issuecomment-2041303349
@fredia thank for your suggestions!
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1555239594
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateExecutor.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.annotation.Internal;
+
+import java.util.concurrent.CompletableFuture;
+
+/** Executor for executing batch {@link StateRequest}s. */
+@Internal
+public interface StateExecutor {
+ /**
+ * Execute a batch of state requests.
+ *
+ * @param processingRequests the given batch of processing requests
+ * @return A future can determine whether execution has completed.
+ */
+ CompletableFuture<Boolean> executeBatchRequests(
+ Iterable<StateRequest<?, ?, ?>> processingRequests);
Review Comment:
The sorting and batching in group is finished by the `AEC`, there is no mixed types of requests here.
Here we just follow the description of the FLIPs, if we need to change the signature here, it would be fine change this later.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1555237399
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
+
+ /** The type of processing request. */
+ public enum RequestType {
+ /** Process one record without state access. */
+ SYNC,
Review Comment:
It actually means sync with the framework to check if there is another record on-going with a same key. It is something like `SYNC_POINT`. I'll change the doc 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
Re: [PR] [FLINK-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on PR #24614:
URL: https://github.com/apache/flink/pull/24614#issuecomment-2041936516
@masteryhx Thanks for your comments! Applied some suggestions, PTAL thanks.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1555343181
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/KeyAccountingUnit.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.runtime.asyncprocessing;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Key accounting unit holds the current in-flight key and tracks the corresponding ongoing records,
+ * which is used to preserve the ordering of independent chained {@link
+ * org.apache.flink.api.common.state.v2.StateFuture}.
+ *
+ * @param <R> the type of record
+ * @param <K> the type of key
+ */
+public class KeyAccountingUnit<R, K> {
Review Comment:
Okay, I don't insist this.
Let's consider it when we have a specific case 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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1555244839
##########
flink-runtime/src/test/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionControllerTest.java:
##########
@@ -0,0 +1,305 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.StateFuture;
+import org.apache.flink.api.common.state.v2.ValueState;
+import org.apache.flink.core.state.StateFutureUtils;
+import org.apache.flink.runtime.asyncprocessing.StateRequest.RequestType;
+import org.apache.flink.runtime.mailbox.SyncMailboxExecutor;
+import org.apache.flink.util.Preconditions;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.LinkedList;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.assertj.core.api.AssertionsForClassTypes.assertThat;
+
+/** Test for {@link AsyncExecutionController}. */
+class AsyncExecutionControllerTest {
+
+ // TODO: this test is not well completed, cause buffering in AEC is not implemented.
+ // Yet, just for illustrating the interaction between AEC and Async state API.
+ @Test
+ void testBasicRun() {
+ TestAsyncExecutionController<String, String> aec =
+ new TestAsyncExecutionController<>(
+ new SyncMailboxExecutor(), new TestStateExecutor());
+ TestUnderlyingState underlyingState = new TestUnderlyingState();
+ TestValueState valueState = new TestValueState(aec, underlyingState);
+ AtomicInteger output = new AtomicInteger();
+ Runnable userCode =
+ () -> {
+ valueState
+ .asyncValue()
+ .thenCompose(
+ val -> {
+ int updated = (val == null ? 1 : (val + 1));
+ return valueState
+ .asyncUpdate(updated)
+ .thenCompose(
+ o ->
+ StateFutureUtils.completedFuture(
+ updated));
+ })
+ .thenAccept(val -> output.set(val));
+ };
+
+ // ============================ element1 ============================
+ String record1 = "key1-r1";
+ String key1 = "key1";
+ // Simulate the wrapping in {@link RecordProcessorUtils#getRecordProcessor()}, wrapping the
+ // record and key with RecordContext.
+ RecordContext<String, String> recordContext1 = aec.buildContext(record1, key1);
+ aec.setCurrentContext(recordContext1);
+ // execute user code
+ userCode.run();
+
+ // Single-step run.
+ // Firstly, the user code generates value get in active buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // After running, the value update is in active buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // Value update finishes.
+ assertThat(aec.activeBuffer.size()).isEqualTo(0);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(0);
+ assertThat(output.get()).isEqualTo(1);
+ assertThat(recordContext1.getReferenceCount()).isEqualTo(0);
+
+ // ============================ element 2 & 3 ============================
+ String record2 = "key1-r2";
+ String key2 = "key1";
+ RecordContext<String, String> recordContext2 = aec.buildContext(record2, key2);
+ aec.setCurrentContext(recordContext2);
+ // execute user code
+ userCode.run();
+
+ String record3 = "key1-r3";
+ String key3 = "key1";
+ RecordContext<String, String> recordContext3 = aec.buildContext(record3, key3);
+ aec.setCurrentContext(recordContext3);
+ // execute user code
+ userCode.run();
+
+ // Single-step run.
+ // Firstly, the user code for record2 generates value get in active buffer,
+ // while user code for record3 generates value get in blocking buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.blockingBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // After running, the value update for record2 is in active buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.blockingBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // Value update for record2 finishes. The value get for record3 is still in blocking status.
+ assertThat(aec.activeBuffer.size()).isEqualTo(0);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(0);
+ assertThat(output.get()).isEqualTo(2);
+ assertThat(recordContext2.getReferenceCount()).isEqualTo(0);
+ assertThat(aec.blockingBuffer.size()).isEqualTo(1);
+
+ aec.migrateBlockingToActive();
+ // Value get for record3 is ready for run.
+
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.blockingBuffer.size()).isEqualTo(0);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // After running, the value update for record3 is in active buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // Value update for record3 finishes.
+ assertThat(aec.activeBuffer.size()).isEqualTo(0);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(0);
+ assertThat(output.get()).isEqualTo(3);
+ assertThat(recordContext3.getReferenceCount()).isEqualTo(0);
+
+ // ============================ element4 ============================
+ String record4 = "key3-r3";
+ String key4 = "key3";
+ RecordContext<String, String> recordContext4 = aec.buildContext(record4, key4);
+ aec.setCurrentContext(recordContext4);
+ // execute user code
+ userCode.run();
+
+ // Single-step run for another key.
+ // Firstly, the user code generates value get in active buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // After running, the value update is in active buffer.
+ assertThat(aec.activeBuffer.size()).isEqualTo(1);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(1);
+ aec.triggerIfNeeded(true);
+ // Value update finishes.
+ assertThat(aec.activeBuffer.size()).isEqualTo(0);
+ assertThat(aec.keyAccountingUnit.occupiedCount()).isEqualTo(0);
+ assertThat(output.get()).isEqualTo(1);
+ assertThat(recordContext4.getReferenceCount()).isEqualTo(0);
+ }
+
+ /**
+ * An AsyncExecutionController for testing purpose, which integrates with basic buffer
+ * mechanism.
+ */
+ static class TestAsyncExecutionController<R, K> extends AsyncExecutionController<R, K> {
+
+ LinkedList<StateRequest<K, ?, ?>> activeBuffer;
+
+ LinkedList<StateRequest<K, ?, ?>> blockingBuffer;
+
+ public TestAsyncExecutionController(
+ MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) {
+ super(mailboxExecutor, stateExecutor);
+ activeBuffer = new LinkedList<>();
+ blockingBuffer = new LinkedList<>();
+ }
+
+ @Override
+ <IN, OUT> void insertActiveBuffer(StateRequest<K, IN, OUT> request) {
+ activeBuffer.push(request);
+ }
+
+ <IN, OUT> void insertBlockingBuffer(StateRequest<K, IN, OUT> request) {
+ blockingBuffer.push(request);
+ }
+
+ void triggerIfNeeded(boolean force) {
+ if (!force) {
+ // Disable normal trigger, to perform single-step debugging and check.
+ return;
+ }
+ LinkedList<StateRequest<?, ?, ?>> toRun = new LinkedList<>(activeBuffer);
+ activeBuffer.clear();
+ stateExecutor.executeBatchRequests(toRun);
+ }
+
+ void migrateBlockingToActive() {
+ Iterator<StateRequest<K, ?, ?>> blockingIter = blockingBuffer.iterator();
+ while (blockingIter.hasNext()) {
+ StateRequest<K, ?, ?> request = blockingIter.next();
+ if (request.getRecordContext().tryOccupyKey()) {
+ insertActiveBuffer(request);
+ blockingIter.remove();
+ }
+ }
+ }
+ }
+
+ /** Simulate the underlying state that is actually used to execute the request. */
+ static class TestUnderlyingState {
+
+ private HashMap<String, Integer> hashMap;
+
+ public TestUnderlyingState() {
+ this.hashMap = new HashMap<>();
+ }
+
+ public Integer get(String key) {
+ return hashMap.get(key);
+ }
+
+ public void update(String key, Integer val) {
+ hashMap.put(key, val);
+ }
+ }
+
+ static class TestValueState implements ValueState<Integer> {
+
+ private AsyncExecutionController<String, String> asyncExecutionController;
+
+ private TestUnderlyingState underlyingState;
+
+ public TestValueState(
+ AsyncExecutionController<String, String> aec, TestUnderlyingState underlyingState) {
+ this.asyncExecutionController = aec;
+ this.underlyingState = underlyingState;
+ }
+
+ @Override
+ public StateFuture<Void> asyncClear() {
+ StateRequest<String, Void, Void> request =
Review Comment:
`SYNC` is special case that will be only generated by the framework (only before the `processElement`). The direct implementation of State API should not create this kind of request.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1554800173
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/ContextStateFutureImpl.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.core.state.StateFutureImpl;
+
+/**
+ * A state future that holds the {@link RecordContext} and maintains the reference count of it. The
+ * reason why we maintain the reference here is that the ContextStateFutureImpl can be created
+ * multiple times since user may chain their code wildly, some of which are only for internal usage
+ * (See {@link StateFutureImpl}). So maintaining reference counting by the lifecycle of state future
+ * is relatively simple and less error-prone.
+ *
+ * <p>Reference counting added on {@link RecordContext} follows:
+ * <li>1. +1 when this future created.
+ * <li>2. -1 when future completed.
+ * <li>3. +1 when callback registered.
+ * <li>4. -1 when callback finished.
+ */
+public class ContextStateFutureImpl<T> extends StateFutureImpl<T> {
+
+ private final RecordContext<?, ?> recordContext;
+
+ ContextStateFutureImpl(CallbackRunner callbackRunner, RecordContext<?, ?> recordContext) {
+ super(callbackRunner);
+ this.recordContext = recordContext;
+ // When state request submitted, ref count +1, as described in FLIP-425:
+ // To cover the statements without a callback, in addition to the reference count marked
+ // in Fig.5, each state request itself is also protected by a paired reference count.
+ recordContext.retain();
+ }
+
+ @Override
+ public <A> StateFutureImpl<A> makeNewStateFuture() {
+ return new ContextStateFutureImpl<>(callbackRunner, recordContext);
+ }
+
+ @Override
+ public void callbackRegistered() {
+ // When a callback registered, as shown in Fig.5 of FLIP-425, at the point of 3 and 5, the
+ // ref count -1.
Review Comment:
+1?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
+
+ /** The type of processing request. */
+ public enum RequestType {
+ /** Process one record without state access. */
+ SYNC,
+ /** Get from one {@link State}. */
+ GET,
+ /** Put to one {@link State}. */
+ PUT,
+ /** Merge value to an exist key in {@link State}. Mainly used for listState. */
+ MERGE,
+ /** Delete from one {@link State}. */
+ DELETE
+ }
+
+ /** The underlying state to be accessed, can be empty for {@link RequestType#SYNC}. */
+ @Nullable private final State state;
Review Comment:
Should the type of `state` be restricted to `org.apache.flink.api.common.state.v2.State`?
IIUC, this underlying state is a lower-level state that directly interacts with the KV engine(e.g. ForSt/RocksDB) , and `org.apache.flink.api.common.state.v2.State` is a user-level interface. WDYT?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/ContextStateFutureImpl.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.core.state.StateFutureImpl;
+
+/**
+ * A state future that holds the {@link RecordContext} and maintains the reference count of it. The
+ * reason why we maintain the reference here is that the ContextStateFutureImpl can be created
+ * multiple times since user may chain their code wildly, some of which are only for internal usage
+ * (See {@link StateFutureImpl}). So maintaining reference counting by the lifecycle of state future
+ * is relatively simple and less error-prone.
+ *
+ * <p>Reference counting added on {@link RecordContext} follows:
+ * <li>1. +1 when this future created.
+ * <li>2. -1 when future completed.
+ * <li>3. +1 when callback registered.
+ * <li>4. -1 when callback finished.
+ */
+public class ContextStateFutureImpl<T> extends StateFutureImpl<T> {
+
+ private final RecordContext<?, ?> recordContext;
+
+ ContextStateFutureImpl(CallbackRunner callbackRunner, RecordContext<?, ?> recordContext) {
+ super(callbackRunner);
+ this.recordContext = recordContext;
+ // When state request submitted, ref count +1, as described in FLIP-425:
+ // To cover the statements without a callback, in addition to the reference count marked
+ // in Fig.5, each state request itself is also protected by a paired reference count.
+ recordContext.retain();
+ }
+
+ @Override
+ public <A> StateFutureImpl<A> makeNewStateFuture() {
+ return new ContextStateFutureImpl<>(callbackRunner, recordContext);
+ }
+
+ @Override
+ public void callbackRegistered() {
+ // When a callback registered, as shown in Fig.5 of FLIP-425, at the point of 3 and 5, the
+ // ref count -1.
+ recordContext.retain();
+ }
+
+ @Override
+ public void postComplete() {
Review Comment:
-1?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateFutureFactory.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.core.state.InternalStateFuture;
+
+/**
+ * An internal factory for {@link InternalStateFuture} that build future with necessary context
+ * switch and wired with mailbox executor.
+ */
+public class StateFutureFactory<R, K> {
Review Comment:
Mark `@Internal`?
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1558680967
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
+
+ /** The type of processing request. */
+ public enum RequestType {
+ /** Process one record without state access. */
+ SYNC,
+ /** Get from one {@link State}. */
+ GET,
+ /** Put to one {@link State}. */
+ PUT,
+ /** Merge value to an exist key in {@link State}. Mainly used for listState. */
+ MERGE,
+ /** Delete from one {@link State}. */
+ DELETE
+ }
+
+ /** The underlying state to be accessed, can be empty for {@link RequestType#SYNC}. */
+ @Nullable private final State state;
+
+ /** The type of this request. */
+ private final RequestType type;
+
+ /** The payload(input) of this request. */
+ @Nullable private final IN payload;
+
+ /** The future to collect the result of the request. */
+ private InternalStateFuture<OUT> stateFuture;
+
+ /** The record context of this request. */
+ private RecordContext<?, K> context;
+
+ StateRequest(@Nullable State state, RequestType type, @Nullable IN payload) {
+ this.state = state;
+ this.type = type;
+ this.payload = payload;
+ }
+
+ RequestType getRequestType() {
+ return type;
+ }
+
+ @Nullable
+ IN getPayload() {
+ return payload;
+ }
+
+ @Nullable
+ State getState() {
+ return state;
+ }
+
+ InternalStateFuture<OUT> getFuture() {
+ return stateFuture;
+ }
+
+ void setFuture(InternalStateFuture<OUT> future) {
+ stateFuture = future;
+ }
+
+ RecordContext<?, K> getRecordContext() {
+ return context;
+ }
+
+ void setRecordContext(RecordContext<?, K> context) {
+ this.context = context;
+ }
Review Comment:
This LGTM currently
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1554965456
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
+
+ /** The type of processing request. */
+ public enum RequestType {
+ /** Process one record without state access. */
+ SYNC,
+ /** Get from one {@link State}. */
+ GET,
+ /** Put to one {@link State}. */
+ PUT,
+ /** Merge value to an exist key in {@link State}. Mainly used for listState. */
+ MERGE,
+ /** Delete from one {@link State}. */
+ DELETE
+ }
+
+ /** The underlying state to be accessed, can be empty for {@link RequestType#SYNC}. */
+ @Nullable private final State state;
+
+ /** The type of this request. */
+ private final RequestType type;
+
+ /** The payload(input) of this request. */
+ @Nullable private final IN payload;
+
+ /** The future to collect the result of the request. */
+ private InternalStateFuture<OUT> stateFuture;
+
+ /** The record context of this request. */
+ private RecordContext<?, K> context;
+
+ StateRequest(@Nullable State state, RequestType type, @Nullable IN payload) {
+ this.state = state;
+ this.type = type;
+ this.payload = payload;
+ }
+
+ RequestType getRequestType() {
+ return type;
+ }
+
+ @Nullable
+ IN getPayload() {
+ return payload;
+ }
+
+ @Nullable
+ State getState() {
+ return state;
+ }
+
+ InternalStateFuture<OUT> getFuture() {
+ return stateFuture;
+ }
+
+ void setFuture(InternalStateFuture<OUT> future) {
+ stateFuture = future;
+ }
+
+ RecordContext<?, K> getRecordContext() {
+ return context;
+ }
+
+ void setRecordContext(RecordContext<?, K> context) {
+ this.context = context;
+ }
Review Comment:
@ljz2051 The reason behind this is that the `RecordContext` is managed by the `AEC` and related components, while the `StateRequest` is created in direct implementation of State API (`internal key-value layer`), which lacks the knowledge of `RecordContext`. I tried to avoid introducing `getRecordContext` in `AEC` and call this in every implementation of the state interface.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1555253454
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
+
+ /** The type of processing request. */
+ public enum RequestType {
+ /** Process one record without state access. */
+ SYNC,
+ /** Get from one {@link State}. */
+ GET,
+ /** Put to one {@link State}. */
+ PUT,
+ /** Merge value to an exist key in {@link State}. Mainly used for listState. */
+ MERGE,
+ /** Delete from one {@link State}. */
+ DELETE
+ }
+
+ /** The underlying state to be accessed, can be empty for {@link RequestType#SYNC}. */
+ @Nullable private final State state;
+
+ /** The type of this request. */
+ private final RequestType type;
+
+ /** The payload(input) of this request. */
+ @Nullable private final IN payload;
+
+ /** The future to collect the result of the request. */
+ private InternalStateFuture<OUT> stateFuture;
+
+ /** The record context of this request. */
+ private RecordContext<?, K> context;
+
+ StateRequest(@Nullable State state, RequestType type, @Nullable IN payload) {
+ this.state = state;
+ this.type = type;
+ this.payload = payload;
+ }
+
+ RequestType getRequestType() {
+ return type;
+ }
+
+ @Nullable
+ IN getPayload() {
+ return payload;
+ }
+
+ @Nullable
+ State getState() {
+ return state;
+ }
+
+ InternalStateFuture<OUT> getFuture() {
+ return stateFuture;
+ }
+
+ void setFuture(InternalStateFuture<OUT> future) {
+ stateFuture = future;
+ }
+
+ RecordContext<?, K> getRecordContext() {
+ return context;
+ }
+
+ void setRecordContext(RecordContext<?, K> context) {
+ this.context = context;
+ }
Review Comment:
@masteryhx After trying, I think we are still not able to avoid lazy setting some contexts or create some temporary instances. How about we make `AEC#handleRequest` receive the information from outside and build the `StateRequest` internally?
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "jiexray (via GitHub)" <gi...@apache.org>.
jiexray commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1557546908
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
+
+ /** The corresponding context that currently runs in task thread. */
+ RecordContext<R, K> currentContext;
Review Comment:
I am not quite sure about the "fully managed". In my opinion, "fully managed" is equivalent that the context is created/used/destroyed by AEC. Could you give an example -- whose may invoke buildContext() to create a `RecordContext`?
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1554815099
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
+
+ /** The type of processing request. */
+ public enum RequestType {
+ /** Process one record without state access. */
+ SYNC,
+ /** Get from one {@link State}. */
+ GET,
+ /** Put to one {@link State}. */
+ PUT,
+ /** Merge value to an exist key in {@link State}. Mainly used for listState. */
+ MERGE,
+ /** Delete from one {@link State}. */
+ DELETE
+ }
+
+ /** The underlying state to be accessed, can be empty for {@link RequestType#SYNC}. */
+ @Nullable private final State state;
Review Comment:
Thanks for the explanation, agreed👍.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1554810305
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateFutureFactory.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.core.state.InternalStateFuture;
+
+/**
+ * An internal factory for {@link InternalStateFuture} that build future with necessary context
+ * switch and wired with mailbox executor.
+ */
+public class StateFutureFactory<R, K> {
Review Comment:
I'd rather keep it with no annotation, since IIUC only interfaces for internal usage should be annotated. The package `org.apache.flink.runtime` is not for APIs but for implementations.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "flinkbot (via GitHub)" <gi...@apache.org>.
flinkbot commented on PR #24614:
URL: https://github.com/apache/flink/pull/24614#issuecomment-2036689302
<!--
Meta data
{
"version" : 1,
"metaDataEntries" : [ {
"hash" : "fee30d9e3d135c9a4c32e544b8165e8c93d5c1e6",
"status" : "UNKNOWN",
"url" : "TBD",
"triggerID" : "fee30d9e3d135c9a4c32e544b8165e8c93d5c1e6",
"triggerType" : "PUSH"
} ]
}-->
## CI report:
* fee30d9e3d135c9a4c32e544b8165e8c93d5c1e6 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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1555241719
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/KeyAccountingUnit.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.runtime.asyncprocessing;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Key accounting unit holds the current in-flight key and tracks the corresponding ongoing records,
+ * which is used to preserve the ordering of independent chained {@link
+ * org.apache.flink.api.common.state.v2.StateFuture}.
+ *
+ * @param <R> the type of record
+ * @param <K> the type of key
+ */
+public class KeyAccountingUnit<R, K> {
Review Comment:
I'm not a big fan of interface abstraction. Two questions: Is there a need for two or more behaviors? And how could we/user choose one of them? It is just a class for internal usage so I'd keep this normal class until we meet a concrete use case for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "jiexray (via GitHub)" <gi...@apache.org>.
jiexray commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1557531709
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
Review Comment:
`protect KeyAccountingUnit<R, K> keyAccountingUnit` may be better? Only subclass can see this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "jiexray (via GitHub)" <gi...@apache.org>.
jiexray commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1556755468
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
Review Comment:
why is this not private?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
+
+ /** The corresponding context that currently runs in task thread. */
+ RecordContext<R, K> currentContext;
Review Comment:
Could we make `currentContext` as a parameter of the functions in aec, rather than the member of aec.
```
public <IN, OUT> InternalStateFuture<OUT> handleRequest(
@Nullable State state, StateRequest.RequestType type, @Nullable IN payload, RecordContext<R, K> context);
```
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
Review Comment:
Why this is not private?
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1557607902
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
Review Comment:
I'd prefer current status (without any modifier). No modifier (also `package-private`) is more strict than `protected`. See "Access Levels" in https://docs.oracle.com/javase/tutorial/java/javaOO/accesscontrol.html . Currently we won't access this member outside the package.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "jiexray (via GitHub)" <gi...@apache.org>.
jiexray commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1557717884
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
+
+ /** The corresponding context that currently runs in task thread. */
+ RecordContext<R, K> currentContext;
Review Comment:
Thank you for your apply. I think I get a more clear view on the `RecordContext` now. From the testing code, I find the lifecycle of `RecordContext` seems to be
```
buildContext() (named as ctx1) -> aec.setCurrentContext(ctx1) -> user_code ->
async state -> aec.handleRequest() -> new StateRequest(ctx1) -> async state ->
aec.handleRequest() -> new StateRequest(ctx1)
```
This async state operations form a chain, and all operations in the chain use the `ctx1`.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
+
+ /** The corresponding context that currently runs in task thread. */
+ RecordContext<R, K> currentContext;
Review Comment:
Thank you for your apply. I think I get a more clear view on the `RecordContext` now. From the testing code, I find the lifecycle of `RecordContext` seems to be
```
buildContext() (named as ctx1) -> aec.setCurrentContext(ctx1) -> user_code ->
async state -> aec.handleRequest() -> new StateRequest(ctx1) -> async state ->
aec.handleRequest() -> new StateRequest(ctx1)
```
This async state operations form a chain, and all operations in the chain use the `ctx1`.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "fredia (via GitHub)" <gi...@apache.org>.
fredia merged PR #24614:
URL: https://github.com/apache/flink/pull/24614
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "ljz2051 (via GitHub)" <gi...@apache.org>.
ljz2051 commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1554960979
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateRequest.java:
##########
@@ -0,0 +1,99 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import javax.annotation.Nullable;
+
+/**
+ * A request encapsulates the necessary data to perform a state request.
+ *
+ * @param <K> Type of partitioned key.
+ * @param <IN> Type of input of this request.
+ * @param <OUT> Type of value that request will return.
+ */
+public class StateRequest<K, IN, OUT> {
+
+ /** The type of processing request. */
+ public enum RequestType {
+ /** Process one record without state access. */
+ SYNC,
+ /** Get from one {@link State}. */
+ GET,
+ /** Put to one {@link State}. */
+ PUT,
+ /** Merge value to an exist key in {@link State}. Mainly used for listState. */
+ MERGE,
+ /** Delete from one {@link State}. */
+ DELETE
+ }
+
+ /** The underlying state to be accessed, can be empty for {@link RequestType#SYNC}. */
+ @Nullable private final State state;
+
+ /** The type of this request. */
+ private final RequestType type;
+
+ /** The payload(input) of this request. */
+ @Nullable private final IN payload;
+
+ /** The future to collect the result of the request. */
+ private InternalStateFuture<OUT> stateFuture;
+
+ /** The record context of this request. */
+ private RecordContext<?, K> context;
+
+ StateRequest(@Nullable State state, RequestType type, @Nullable IN payload) {
+ this.state = state;
+ this.type = type;
+ this.payload = payload;
+ }
+
+ RequestType getRequestType() {
+ return type;
+ }
+
+ @Nullable
+ IN getPayload() {
+ return payload;
+ }
+
+ @Nullable
+ State getState() {
+ return state;
+ }
+
+ InternalStateFuture<OUT> getFuture() {
+ return stateFuture;
+ }
+
+ void setFuture(InternalStateFuture<OUT> future) {
+ stateFuture = future;
+ }
+
+ RecordContext<?, K> getRecordContext() {
+ return context;
+ }
+
+ void setRecordContext(RecordContext<?, K> context) {
+ this.context = context;
+ }
Review Comment:
The key in RecordContext is essential for stateRequest. Without setting the RecordContext, the stateRequest is essentially "incomplete". Why don't we consider making RecordContext a final constructor variable for stateRequest?
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1556939595
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
+
+ /** The corresponding context that currently runs in task thread. */
+ RecordContext<R, K> currentContext;
Review Comment:
The `currentContext` is fully managed by `AEC` and related runtime components, while the `handleRequest` is called by the direct implementation of `State APIs`, which is somewhat upper layer of `AEC` and lacks the knowledge of this context.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "masteryhx (via GitHub)" <gi...@apache.org>.
masteryhx commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1556826447
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateExecutor.java:
##########
@@ -0,0 +1,36 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.annotation.Internal;
+
+import java.util.concurrent.CompletableFuture;
+
+/** Executor for executing batch {@link StateRequest}s. */
+@Internal
+public interface StateExecutor {
+ /**
+ * Execute a batch of state requests.
+ *
+ * @param processingRequests the given batch of processing requests
+ * @return A future can determine whether execution has completed.
+ */
+ CompletableFuture<Boolean> executeBatchRequests(
+ Iterable<StateRequest<?, ?, ?>> processingRequests);
Review Comment:
As discussed offline:
1. We should introduce more request types for executor to generate the real state.
2. An Internal State API in other PR will be introduced to provide a state name, and as a field of StateRequest.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1555245552
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.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.runtime.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
+
+ /** The corresponding context that currently runs in task thread. */
+ RecordContext<R, K> currentContext;
+
+ public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) {
+ this(mailboxExecutor, stateExecutor, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM);
+ }
+
+ public AsyncExecutionController(
+ MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int maxInFlightRecords) {
+ this.keyAccountingUnit = new KeyAccountingUnit<>();
+ this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor);
+ this.stateExecutor = stateExecutor;
+ this.maxInFlightRecordNum = maxInFlightRecords;
+ LOG.info("Create AsyncExecutionController: maxInFlightRecordsNum {}", maxInFlightRecords);
+ }
+
+ /**
+ * Build a new context based on record and key. Also wired with internal {@link
+ * KeyAccountingUnit}.
+ *
+ * @param record the given record.
+ * @param key the given key.
+ * @return the built record context.
+ */
+ public RecordContext<R, K> buildContext(R record, K key) {
Review Comment:
Well, building/initializing is a special use case other than simple setting. We can change the behavior of this method to something like `buildAndSet`. But I slightly tend to make interfaces implement more specialized(single) functions rather than multiple.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1556936793
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
Review Comment:
Same as above.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1556936013
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,144 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
Review Comment:
These are used by the `AsyncExecutionControllerTest#TestAsyncExecutionController`. Since the AEC is incomplete in some way, we perform test based on its testing subclasses.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1559205982
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
+
+ /** The corresponding context that currently runs in task thread. */
+ RecordContext<R, K> currentContext;
+
+ public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) {
+ this(mailboxExecutor, stateExecutor, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM);
+ }
+
+ public AsyncExecutionController(
+ MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int maxInFlightRecords) {
+ this.keyAccountingUnit = new KeyAccountingUnit<>();
+ this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor);
+ this.stateExecutor = stateExecutor;
+ this.maxInFlightRecordNum = maxInFlightRecords;
+ LOG.info("Create AsyncExecutionController: maxInFlightRecordsNum {}", maxInFlightRecords);
+ }
+
+ /**
+ * Build a new context based on record and key. Also wired with internal {@link
+ * KeyAccountingUnit}.
+ *
+ * @param record the given record.
+ * @param key the given key.
+ * @return the built record context.
+ */
+ public RecordContext<R, K> buildContext(R record, K key) {
+ return new RecordContext<>(record, key, this::disposeContext);
+ }
+
+ /**
+ * Each time before a code segment (callback) is about to run in mailbox (task thread), this
+ * method should be called to switch a context in AEC.
+ *
+ * @param switchingContext the context to switch.
+ */
+ public void setCurrentContext(RecordContext<R, K> switchingContext) {
+ currentContext = switchingContext;
+ }
+
+ /**
+ * Dispose a context.
+ *
+ * @param toDispose the context to dispose.
+ */
+ public void disposeContext(RecordContext<R, K> toDispose) {
+ keyAccountingUnit.release(toDispose.getRecord(), toDispose.getKey());
+ }
+
+ /**
+ * Try to occupy a key by a given context.
+ *
+ * @param recordContext the given context.
+ * @return true if occupy succeed or the key has already occupied by this context.
+ */
+ boolean tryOccupyKey(RecordContext<R, K> recordContext) {
+ boolean occupied = recordContext.isKeyOccupied();
+ if (!occupied
+ && keyAccountingUnit.occupy(recordContext.getRecord(), recordContext.getKey())) {
+ recordContext.setKeyOccupied();
+ occupied = true;
+ }
+ return occupied;
+ }
+
+ /**
+ * Submit a {@link StateRequest} to this AEC and trigger if needed.
+ *
+ * @param state the state to request.
+ * @param type the type of this request.
+ * @param payload the payload input for this request.
+ * @return the state future.
+ */
+ public <IN, OUT> InternalStateFuture<OUT> handleRequest(
+ @Nullable State state, StateRequestType type, @Nullable IN payload) {
Review Comment:
Could you please remind me of the cases when a `null` state would be passed to AEC?
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/KeyAccountingUnit.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.runtime.asyncprocessing;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Key accounting unit holds the current in-flight key and tracks the corresponding ongoing records,
+ * which is used to preserve the ordering of independent chained {@link
+ * org.apache.flink.api.common.state.v2.StateFuture}.
+ *
+ * @param <R> the type of record
+ * @param <K> the type of key
+ */
+public class KeyAccountingUnit<R, K> {
+
+ /** The in-flight records that are being processed, their keys are different from each other. */
+ private final Map<K, R> noConflictInFlightRecords;
+
+ public KeyAccountingUnit() {
+ this.noConflictInFlightRecords = new ConcurrentHashMap<>();
Review Comment:
Given that DEFAULT_MAX_IN_FLIGHT_RECORD_NUM is pre-determined, we can set the initial capacity of this map for potentially reducing memory space.
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
Review Comment:
nit: allow -> allowed
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/StateFutureFactory.java:
##########
@@ -0,0 +1,51 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.core.state.InternalStateFuture;
+
+/**
+ * An internal factory for {@link InternalStateFuture} that build future with necessary context
+ * switch and wired with mailbox executor.
+ */
+public class StateFutureFactory<R, K> {
+
+ private final AsyncExecutionController<R, K> asyncExecutionController;
+ private final MailboxExecutor mailboxExecutor;
+
+ StateFutureFactory(
+ AsyncExecutionController<R, K> asyncExecutionController,
+ MailboxExecutor mailboxExecutor) {
+ this.asyncExecutionController = asyncExecutionController;
+ this.mailboxExecutor = mailboxExecutor;
+ }
+
+ public <OUT> InternalStateFuture<OUT> build(RecordContext<R, K> context) {
Review Comment:
nit: it might be better to change the method name to "create".
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "yunfengzhou-hub (via GitHub)" <gi...@apache.org>.
yunfengzhou-hub commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1559185609
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/KeyAccountingUnit.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.runtime.asyncprocessing;
+
+import org.apache.flink.annotation.VisibleForTesting;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Key accounting unit holds the current in-flight key and tracks the corresponding ongoing records,
+ * which is used to preserve the ordering of independent chained {@link
+ * org.apache.flink.api.common.state.v2.StateFuture}.
+ *
+ * @param <R> the type of record
+ * @param <K> the type of key
+ */
+public class KeyAccountingUnit<R, K> {
+
+ /** The in-flight records that are being processed, their keys are different from each other. */
+ private final Map<K, R> noConflictInFlightRecords;
+
+ public KeyAccountingUnit() {
+ this.noConflictInFlightRecords = new ConcurrentHashMap<>();
Review Comment:
Given that maxInFlightRecordNum is determined, we can set the initial capacity of this map for potentially reducing memory space.
--
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-34986][Runtime/State] Basic framework of async execution for state [flink]
Posted by "Zakelly (via GitHub)" <gi...@apache.org>.
Zakelly commented on code in PR #24614:
URL: https://github.com/apache/flink/pull/24614#discussion_r1559350216
##########
flink-runtime/src/main/java/org/apache/flink/runtime/asyncprocessing/AsyncExecutionController.java:
##########
@@ -0,0 +1,169 @@
+/*
+ * 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.asyncprocessing;
+
+import org.apache.flink.api.common.operators.MailboxExecutor;
+import org.apache.flink.api.common.state.v2.State;
+import org.apache.flink.core.state.InternalStateFuture;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nullable;
+
+/**
+ * The Async Execution Controller (AEC) receives processing requests from operators, and put them
+ * into execution according to some strategies.
+ *
+ * <p>It is responsible for:
+ * <li>Preserving the sequence of elements bearing the same key by delaying subsequent requests
+ * until the processing of preceding ones is finalized.
+ * <li>Tracking the in-flight data(records) and blocking the input if too much data in flight
+ * (back-pressure). It invokes {@link MailboxExecutor#yield()} to pause current operations,
+ * allowing for the execution of callbacks (mails in Mailbox).
+ *
+ * @param <R> the type of the record
+ * @param <K> the type of the key
+ */
+public class AsyncExecutionController<R, K> {
+
+ private static final Logger LOG = LoggerFactory.getLogger(AsyncExecutionController.class);
+
+ public static final int DEFAULT_MAX_IN_FLIGHT_RECORD_NUM = 6000;
+
+ /** The max allow number of in-flight records. */
+ private final int maxInFlightRecordNum;
+
+ /** The key accounting unit which is used to detect the key conflict. */
+ final KeyAccountingUnit<R, K> keyAccountingUnit;
+
+ /**
+ * A factory to build {@link org.apache.flink.core.state.InternalStateFuture}, this will auto
+ * wire the created future with mailbox executor. Also conducting the context switch.
+ */
+ private final StateFutureFactory<R, K> stateFutureFactory;
+
+ /** The state executor where the {@link StateRequest} is actually executed. */
+ final StateExecutor stateExecutor;
+
+ /** The corresponding context that currently runs in task thread. */
+ RecordContext<R, K> currentContext;
+
+ public AsyncExecutionController(MailboxExecutor mailboxExecutor, StateExecutor stateExecutor) {
+ this(mailboxExecutor, stateExecutor, DEFAULT_MAX_IN_FLIGHT_RECORD_NUM);
+ }
+
+ public AsyncExecutionController(
+ MailboxExecutor mailboxExecutor, StateExecutor stateExecutor, int maxInFlightRecords) {
+ this.keyAccountingUnit = new KeyAccountingUnit<>();
+ this.stateFutureFactory = new StateFutureFactory<>(this, mailboxExecutor);
+ this.stateExecutor = stateExecutor;
+ this.maxInFlightRecordNum = maxInFlightRecords;
+ LOG.info("Create AsyncExecutionController: maxInFlightRecordsNum {}", maxInFlightRecords);
+ }
+
+ /**
+ * Build a new context based on record and key. Also wired with internal {@link
+ * KeyAccountingUnit}.
+ *
+ * @param record the given record.
+ * @param key the given key.
+ * @return the built record context.
+ */
+ public RecordContext<R, K> buildContext(R record, K key) {
+ return new RecordContext<>(record, key, this::disposeContext);
+ }
+
+ /**
+ * Each time before a code segment (callback) is about to run in mailbox (task thread), this
+ * method should be called to switch a context in AEC.
+ *
+ * @param switchingContext the context to switch.
+ */
+ public void setCurrentContext(RecordContext<R, K> switchingContext) {
+ currentContext = switchingContext;
+ }
+
+ /**
+ * Dispose a context.
+ *
+ * @param toDispose the context to dispose.
+ */
+ public void disposeContext(RecordContext<R, K> toDispose) {
+ keyAccountingUnit.release(toDispose.getRecord(), toDispose.getKey());
+ }
+
+ /**
+ * Try to occupy a key by a given context.
+ *
+ * @param recordContext the given context.
+ * @return true if occupy succeed or the key has already occupied by this context.
+ */
+ boolean tryOccupyKey(RecordContext<R, K> recordContext) {
+ boolean occupied = recordContext.isKeyOccupied();
+ if (!occupied
+ && keyAccountingUnit.occupy(recordContext.getRecord(), recordContext.getKey())) {
+ recordContext.setKeyOccupied();
+ occupied = true;
+ }
+ return occupied;
+ }
+
+ /**
+ * Submit a {@link StateRequest} to this AEC and trigger if needed.
+ *
+ * @param state the state to request.
+ * @param type the type of this request.
+ * @param payload the payload input for this request.
+ * @return the state future.
+ */
+ public <IN, OUT> InternalStateFuture<OUT> handleRequest(
+ @Nullable State state, StateRequestType type, @Nullable IN payload) {
Review Comment:
Added some description in javadoc. This is for the strict order of 'processElement' for same-key records.
--
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