You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by yh...@apache.org on 2023/06/21 15:38:10 UTC
[beam] branch master updated: [runners-flink] #27175 fix coder for eagerly bound state (#27176)
This is an automated email from the ASF dual-hosted git repository.
yhu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 5f7ff39fd06 [runners-flink] #27175 fix coder for eagerly bound state (#27176)
5f7ff39fd06 is described below
commit 5f7ff39fd062e45b59875324d6d06ac3c006f31d
Author: Jan Lukavský <je...@seznam.cz>
AuthorDate: Wed Jun 21 17:38:03 2023 +0200
[runners-flink] #27175 fix coder for eagerly bound state (#27176)
* [runners-flink] #27175 fix coder for eagerly bound state
* [website] update pipeline options for FlinkRunner
---
.../translation/wrappers/streaming/state/FlinkStateInternals.java | 7 ++++---
.../www/site/layouts/shortcodes/flink_java_pipeline_options.html | 5 +++++
.../www/site/layouts/shortcodes/flink_python_pipeline_options.html | 5 +++++
3 files changed, 14 insertions(+), 3 deletions(-)
diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java
index 16a28e8ce67..5c70bc39799 100644
--- a/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java
+++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/state/FlinkStateInternals.java
@@ -76,7 +76,6 @@ import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeutils.TypeSerializer;
import org.apache.flink.api.common.typeutils.base.BooleanSerializer;
import org.apache.flink.api.common.typeutils.base.StringSerializer;
-import org.apache.flink.api.common.typeutils.base.VoidSerializer;
import org.apache.flink.runtime.state.KeyedStateBackend;
import org.apache.flink.runtime.state.VoidNamespace;
import org.apache.flink.runtime.state.VoidNamespaceSerializer;
@@ -1409,7 +1408,9 @@ public class FlinkStateInternals<K> implements StateInternals {
this.flinkStateBackend = flinkStateBackend;
this.flinkStateDescriptor =
new MapStateDescriptor<>(
- stateId, new CoderTypeSerializer<>(coder, pipelineOptions), new BooleanSerializer());
+ stateId,
+ new CoderTypeSerializer<>(coder, pipelineOptions),
+ BooleanSerializer.INSTANCE);
}
@Override
@@ -1612,7 +1613,7 @@ public class FlinkStateInternals<K> implements StateInternals {
new MapStateDescriptor<>(
id,
new CoderTypeSerializer<>(elemCoder, pipelineOptions),
- VoidSerializer.INSTANCE));
+ BooleanSerializer.INSTANCE));
} catch (Exception e) {
throw new RuntimeException(e);
}
diff --git a/website/www/site/layouts/shortcodes/flink_java_pipeline_options.html b/website/www/site/layouts/shortcodes/flink_java_pipeline_options.html
index 8314d5b6879..ba8b597aaee 100644
--- a/website/www/site/layouts/shortcodes/flink_java_pipeline_options.html
+++ b/website/www/site/layouts/shortcodes/flink_java_pipeline_options.html
@@ -132,6 +132,11 @@ Should be called before running the tests.
<td>Sets the behavior of reusing objects.</td>
<td>Default: <code>false</code></td>
</tr>
+<tr>
+ <td><code>operatorChaining</code></td>
+ <td>Sets the behavior of operator chaining.</td>
+ <td>Default: <code>true</code></td>
+</tr>
<tr>
<td><code>parallelism</code></td>
<td>The degree of parallelism to be used when distributing operations onto workers. If the parallelism is not set, the configured Flink default is used, or 1 if none can be found.</td>
diff --git a/website/www/site/layouts/shortcodes/flink_python_pipeline_options.html b/website/www/site/layouts/shortcodes/flink_python_pipeline_options.html
index 010d3c94282..5293f35e6a1 100644
--- a/website/www/site/layouts/shortcodes/flink_python_pipeline_options.html
+++ b/website/www/site/layouts/shortcodes/flink_python_pipeline_options.html
@@ -132,6 +132,11 @@ Should be called before running the tests.
<td>Sets the behavior of reusing objects.</td>
<td>Default: <code>false</code></td>
</tr>
+<tr>
+ <td><code>operator_chaining</code></td>
+ <td>Sets the behavior of operator chaining.</td>
+ <td>Default: <code>true</code></td>
+</tr>
<tr>
<td><code>parallelism</code></td>
<td>The degree of parallelism to be used when distributing operations onto workers. If the parallelism is not set, the configured Flink default is used, or 1 if none can be found.</td>