Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/5239#discussion_r162584893
--- Diff:
flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/StreamOperatorSnapshotRestoreTest.java
---
@@ -98,7 +101,26 @@ public Integer getKey(Integer value) throws Exception {
TypeInformation.of(Integer.class),
MAX_PARALLELISM,
1 /* num subtasks */,
- 0 /* subtask index */);
+ 0 /* subtask index */) {
+
+ @Override
+ protected StreamTaskStateManager
createStreamTaskStateManager(
+ Environment env,
+ StateBackend stateBackend,
+ ProcessingTimeService processingTimeService) {
+
+ return new StreamTaskStateManagerImpl(env,
stateBackend, processingTimeService) {
+ @Override
+ protected <K>
InternalTimeServiceManager<?, K> internalTimeServiceManager(
+ AbstractKeyedStateBackend<K>
keyedStatedBackend,
+ KeyContext keyContext,
+
Iterable<KeyGroupStatePartitionStreamProvider> rawKeyedStates) throws Exception
{
+
+ return null;
--- End diff --
Yes, this is a workaround, because this is a dedicated test for raw keyed
states - which is otherwise consumed immediately to restore a timer service.
This will fail because we wrote only test data to the stream, not the data for
a timer service.
---
