You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by tz...@apache.org on 2020/02/21 15:55:39 UTC
[flink-statefun] 03/03: [FLINK-16063] [config] Add
ASYNC_MAX_OPERATIONS_PER_TASK config
This is an automated email from the ASF dual-hosted git repository.
tzulitai pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-statefun.git
commit 494f988514ddd76d088593e06f789e15cf706a02
Author: Tzu-Li (Gordon) Tai <tz...@apache.org>
AuthorDate: Fri Feb 21 23:42:16 2020 +0800
[FLINK-16063] [config] Add ASYNC_MAX_OPERATIONS_PER_TASK config
---
.../statefun/flink/core/StatefulFunctionsConfig.java | 20 ++++++++++++++++++++
.../flink/core/StatefulFunctionsConfigTest.java | 2 ++
2 files changed, 22 insertions(+)
diff --git a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfig.java b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfig.java
index 04c2c99..688c705 100644
--- a/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfig.java
+++ b/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfig.java
@@ -87,6 +87,13 @@ public class StatefulFunctionsConfig implements Serializable {
.withDescription(
"The number of bytes to use for in memory buffering of the feedback channel, before spilling to disk.");
+ public static final ConfigOption<Integer> ASYNC_MAX_OPERATIONS_PER_TASK =
+ ConfigOptions.key("statefun.async.max-per-task")
+ .intType()
+ .defaultValue(10_000_000)
+ .withDescription(
+ "The max number of async operations per task before backpressure is applied.");
+
/**
* Creates a new {@link StatefulFunctionsConfig} based on the default configurations in the
* current environment set via the {@code flink-conf.yaml}.
@@ -117,6 +124,8 @@ public class StatefulFunctionsConfig implements Serializable {
private MemorySize feedbackBufferSize;
+ private int maxAsyncOperationsPerTask;
+
private Map<String, String> globalConfigurations = new HashMap<>();
/**
@@ -128,6 +137,7 @@ public class StatefulFunctionsConfig implements Serializable {
this.factoryType = configuration.get(USER_MESSAGE_SERIALIZER);
this.flinkJobName = configuration.get(FLINK_JOB_NAME);
this.feedbackBufferSize = configuration.get(TOTAL_MEMORY_USED_FOR_FEEDBACK_CHECKPOINTING);
+ this.maxAsyncOperationsPerTask = configuration.get(ASYNC_MAX_OPERATIONS_PER_TASK);
for (String key : configuration.keySet()) {
if (key.startsWith(MODULE_CONFIG_PREFIX)) {
@@ -173,6 +183,16 @@ public class StatefulFunctionsConfig implements Serializable {
this.feedbackBufferSize = Objects.requireNonNull(size);
}
+ /** Returns the max async operations allowed per task. */
+ public int getMaxAsyncOperationsPerTask() {
+ return maxAsyncOperationsPerTask;
+ }
+
+ /** Sets the max async operations allowed per task. */
+ public void setMaxAsyncOperationsPerTask(int maxAsyncOperationsPerTask) {
+ this.maxAsyncOperationsPerTask = maxAsyncOperationsPerTask;
+ }
+
/**
* Retrieves the universe provider for loading modules.
*
diff --git a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigTest.java b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigTest.java
index 540739e..89629d3 100644
--- a/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigTest.java
+++ b/statefun-flink/statefun-flink-core/src/test/java/org/apache/flink/statefun/flink/core/StatefulFunctionsConfigTest.java
@@ -37,6 +37,7 @@ public class StatefulFunctionsConfigTest {
configuration.set(
StatefulFunctionsConfig.TOTAL_MEMORY_USED_FOR_FEEDBACK_CHECKPOINTING,
MemorySize.ofMebiBytes(100));
+ configuration.set(StatefulFunctionsConfig.ASYNC_MAX_OPERATIONS_PER_TASK, 100);
configuration.setString("statefun.module.global-config.key1", "value1");
configuration.setString("statefun.module.global-config.key2", "value2");
@@ -45,6 +46,7 @@ public class StatefulFunctionsConfigTest {
Assert.assertEquals(stateFunConfig.getFlinkJobName(), testName);
Assert.assertEquals(stateFunConfig.getFactoryType(), MessageFactoryType.WITH_KRYO_PAYLOADS);
Assert.assertEquals(stateFunConfig.getFeedbackBufferSize(), MemorySize.ofMebiBytes(100));
+ Assert.assertEquals(stateFunConfig.getMaxAsyncOperationsPerTask(), 100);
Assert.assertThat(
stateFunConfig.getGlobalConfigurations(), Matchers.hasEntry("key1", "value1"));
Assert.assertThat(