This is an automated email from the ASF dual-hosted git repository.
dwysakowicz pushed a commit to branch release-1.15
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/release-1.15 by this push:
new f19f032daee [FLINK-29645] BatchExecutionKeyedStateBackend is using
incorrect ExecutionConfig when creating serializer
f19f032daee is described below
commit f19f032daee2fabefb4ccc6257740dd491b3a925
Author: Dawid Wysakowicz <[email protected]>
AuthorDate: Mon Oct 17 09:49:57 2022 +0200
[FLINK-29645] BatchExecutionKeyedStateBackend is using incorrect
ExecutionConfig when creating serializer
---
.../state/benchmark/StateBackendBenchmarkUtils.java | 3 ++-
.../sorted/state/BatchExecutionKeyedStateBackend.java | 9 +++++++--
.../sorted/state/BatchExecutionStateBackend.java | 3 ++-
.../state/BatchExecutionInternalTimeServiceTest.java | 19 +++++++++++++------
.../sorted/state/BatchExecutionStateBackendTest.java | 4 +++-
.../BatchExecutionStateBackendVerificationTest.java | 4 +++-
6 files changed, 30 insertions(+), 12 deletions(-)
diff --git
a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark/StateBackendBenchmarkUtils.java
b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark/StateBackendBenchmarkUtils.java
index 377badd844a..13e0c3fef4a 100644
---
a/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark/StateBackendBenchmarkUtils.java
+++
b/flink-state-backends/flink-statebackend-rocksdb/src/test/java/org/apache/flink/contrib/streaming/state/benchmark/StateBackendBenchmarkUtils.java
@@ -36,6 +36,7 @@ import
org.apache.flink.contrib.streaming.state.RocksDBResourceContainer;
import org.apache.flink.core.fs.CloseableRegistry;
import org.apache.flink.core.fs.Path;
import org.apache.flink.metrics.groups.UnregisteredMetricsGroup;
+import org.apache.flink.runtime.operators.testutils.MockEnvironment;
import org.apache.flink.runtime.state.AbstractStateBackend;
import org.apache.flink.runtime.state.CheckpointableKeyedStateBackend;
import org.apache.flink.runtime.state.KeyGroupRange;
@@ -89,7 +90,7 @@ public class StateBackendBenchmarkUtils {
private static CheckpointableKeyedStateBackend<Long>
createBatchExecutionStateBackend() {
return new BatchExecutionStateBackend()
.createKeyedStateBackend(
- null,
+ MockEnvironment.builder().build(),
new JobID(),
"Test",
new LongSerializer(),
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionKeyedStateBackend.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionKeyedStateBackend.java
index a458d4220af..2f63ec430c7 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionKeyedStateBackend.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionKeyedStateBackend.java
@@ -93,10 +93,15 @@ public class BatchExecutionKeyedStateBackend<K> implements
CheckpointableKeyedSt
private final Map<String, KeyGroupedInternalPriorityQueue<?>>
priorityQueues = new HashMap<>();
private final KeyGroupRange keyGroupRange;
+ private final ExecutionConfig executionConfig;
+
public BatchExecutionKeyedStateBackend(
- TypeSerializer<K> keySerializer, KeyGroupRange keyGroupRange) {
+ TypeSerializer<K> keySerializer,
+ KeyGroupRange keyGroupRange,
+ ExecutionConfig executionConfig) {
this.keySerializer = keySerializer;
this.keyGroupRange = keyGroupRange;
+ this.executionConfig = executionConfig;
}
@Override
@@ -167,7 +172,7 @@ public class BatchExecutionKeyedStateBackend<K> implements
CheckpointableKeyedSt
+ "This operation cannot use partitioned state.");
if (!stateDescriptor.isSerializerInitialized()) {
- stateDescriptor.initializeSerializerUnlessSet(new
ExecutionConfig());
+ stateDescriptor.initializeSerializerUnlessSet(executionConfig);
}
State state = states.get(stateDescriptor.getName());
diff --git
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackend.java
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackend.java
index 1542d7bd057..548eec088c6 100644
---
a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackend.java
+++
b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackend.java
@@ -53,7 +53,8 @@ public class BatchExecutionStateBackend implements
StateBackend {
MetricGroup metricGroup,
@Nonnull Collection<KeyedStateHandle> stateHandles,
CloseableRegistry cancelStreamRegistry) {
- return new BatchExecutionKeyedStateBackend<>(keySerializer,
keyGroupRange);
+ return new BatchExecutionKeyedStateBackend<>(
+ keySerializer, keyGroupRange, env.getExecutionConfig());
}
@Override
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceTest.java
index ee8cd76c000..a7703286bdf 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.api.operators.sorted.state;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.api.common.typeutils.base.IntSerializer;
import org.apache.flink.core.fs.CloseableRegistry;
@@ -125,7 +126,8 @@ public class BatchExecutionInternalTimeServiceTest extends
TestLogger {
@Test
public void testFiringEventTimeTimers() throws Exception {
BatchExecutionKeyedStateBackend<Integer> keyedStatedBackend =
- new BatchExecutionKeyedStateBackend<>(KEY_SERIALIZER, new
KeyGroupRange(0, 1));
+ new BatchExecutionKeyedStateBackend<>(
+ KEY_SERIALIZER, new KeyGroupRange(0, 1), new
ExecutionConfig());
InternalTimeServiceManager<Integer> timeServiceManager =
BatchExecutionInternalTimeServiceManager.create(
keyedStatedBackend,
@@ -159,7 +161,8 @@ public class BatchExecutionInternalTimeServiceTest extends
TestLogger {
@Test
public void testSettingSameKeyDoesNotFireTimers() {
BatchExecutionKeyedStateBackend<Integer> keyedStatedBackend =
- new BatchExecutionKeyedStateBackend<>(KEY_SERIALIZER, new
KeyGroupRange(0, 1));
+ new BatchExecutionKeyedStateBackend<>(
+ KEY_SERIALIZER, new KeyGroupRange(0, 1), new
ExecutionConfig());
InternalTimeServiceManager<Integer> timeServiceManager =
BatchExecutionInternalTimeServiceManager.create(
keyedStatedBackend,
@@ -186,7 +189,8 @@ public class BatchExecutionInternalTimeServiceTest extends
TestLogger {
@Test
public void testCurrentWatermark() throws Exception {
BatchExecutionKeyedStateBackend<Integer> keyedStatedBackend =
- new BatchExecutionKeyedStateBackend<>(KEY_SERIALIZER, new
KeyGroupRange(0, 1));
+ new BatchExecutionKeyedStateBackend<>(
+ KEY_SERIALIZER, new KeyGroupRange(0, 1), new
ExecutionConfig());
InternalTimeServiceManager<Integer> timeServiceManager =
BatchExecutionInternalTimeServiceManager.create(
keyedStatedBackend,
@@ -230,7 +234,8 @@ public class BatchExecutionInternalTimeServiceTest extends
TestLogger {
@Test
public void testProcessingTimeTimers() {
BatchExecutionKeyedStateBackend<Integer> keyedStatedBackend =
- new BatchExecutionKeyedStateBackend<>(KEY_SERIALIZER, new
KeyGroupRange(0, 1));
+ new BatchExecutionKeyedStateBackend<>(
+ KEY_SERIALIZER, new KeyGroupRange(0, 1), new
ExecutionConfig());
TestProcessingTimeService processingTimeService = new
TestProcessingTimeService();
InternalTimeServiceManager<Integer> timeServiceManager =
BatchExecutionInternalTimeServiceManager.create(
@@ -263,7 +268,8 @@ public class BatchExecutionInternalTimeServiceTest extends
TestLogger {
@Test
public void testIgnoringEventTimeTimersFromWithinCallback() {
BatchExecutionKeyedStateBackend<Integer> keyedStatedBackend =
- new BatchExecutionKeyedStateBackend<>(KEY_SERIALIZER, new
KeyGroupRange(0, 1));
+ new BatchExecutionKeyedStateBackend<>(
+ KEY_SERIALIZER, new KeyGroupRange(0, 1), new
ExecutionConfig());
TestProcessingTimeService processingTimeService = new
TestProcessingTimeService();
InternalTimeServiceManager<Integer> timeServiceManager =
BatchExecutionInternalTimeServiceManager.create(
@@ -301,7 +307,8 @@ public class BatchExecutionInternalTimeServiceTest extends
TestLogger {
@Test
public void testIgnoringProcessingTimeTimersFromWithinCallback() {
BatchExecutionKeyedStateBackend<Integer> keyedStatedBackend =
- new BatchExecutionKeyedStateBackend<>(KEY_SERIALIZER, new
KeyGroupRange(0, 1));
+ new BatchExecutionKeyedStateBackend<>(
+ KEY_SERIALIZER, new KeyGroupRange(0, 1), new
ExecutionConfig());
TestProcessingTimeService processingTimeService = new
TestProcessingTimeService();
InternalTimeServiceManager<Integer> timeServiceManager =
BatchExecutionInternalTimeServiceManager.create(
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackendTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackendTest.java
index 5c0c2f76fbd..86ccb2a26c6 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackendTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackendTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.api.operators.sorted.state;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.AggregatingState;
@@ -80,7 +81,8 @@ public class BatchExecutionStateBackendTest extends
TestLogger {
private <K> CheckpointableKeyedStateBackend<K> createKeyedBackend(
TypeSerializer<K> keySerializer) {
- return new BatchExecutionKeyedStateBackend<>(keySerializer, new
KeyGroupRange(0, 9));
+ return new BatchExecutionKeyedStateBackend<>(
+ keySerializer, new KeyGroupRange(0, 9), new ExecutionConfig());
}
/**
diff --git
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackendVerificationTest.java
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackendVerificationTest.java
index 999d201ac88..67062bcf75d 100644
---
a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackendVerificationTest.java
+++
b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackendVerificationTest.java
@@ -18,6 +18,7 @@
package org.apache.flink.streaming.api.operators.sorted.state;
+import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.typeutils.base.LongSerializer;
import org.apache.flink.runtime.checkpoint.CheckpointOptions;
import org.apache.flink.runtime.state.CheckpointStreamFactory;
@@ -45,7 +46,8 @@ public class BatchExecutionStateBackendVerificationTest
extends TestLogger {
expectedException.expectMessage("Snapshotting is not supported in
BATCH runtime mode.");
BatchExecutionKeyedStateBackend<Long> stateBackend =
- new BatchExecutionKeyedStateBackend<>(LONG_SERIALIZER, new
KeyGroupRange(0, 9));
+ new BatchExecutionKeyedStateBackend<>(
+ LONG_SERIALIZER, new KeyGroupRange(0, 9), new
ExecutionConfig());
long checkpointId = 0L;
CheckpointStreamFactory streamFactory = new
MemCheckpointStreamFactory(10);
