This is an automated email from the ASF dual-hosted git repository. pnowojski pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
commit 4f7d11117a25b4de5a9574ebfe62b61397a2cb68 Author: Piotr Nowojski <[email protected]> AuthorDate: Wed Oct 28 19:20:34 2026 +0100 [FLINK-41256][runtime] Add moveToNextKey to KeyContext and KeyedStateBackend BatchExecutionKeyedStateBackend processes keys one at a time in sorted order: every key change fires the outgoing key's timers and clears its state, so the current key may only advance with the sorted input. To tell such key changes apart from an operator calling setCurrentKey itself, add KeyContext#moveToNextKey and KeyedStateBackend#moveToNextKey: move to the next key of a key-sorted input, never back to a previous one. They default to setCurrentKey, so only the batch backend implements them differently. AbstractStreamOperator(V2) forward moveToNextKey to the keyed state backend via StreamOperatorStateHandler. Nothing calls moveToNextKey yet, and the batch backend's setCurrentKey still changes the key. Tests of the batch backend and timer services now change keys via moveToNextKey. Generated-by: Claude Code (Claude Opus 5.5) --- .../flink/runtime/state/KeyedStateBackend.java | 9 ++ .../api/operators/AbstractStreamOperator.java | 11 ++ .../api/operators/AbstractStreamOperatorV2.java | 11 ++ .../flink/streaming/api/operators/KeyContext.java | 9 ++ .../api/operators/StreamOperatorStateHandler.java | 16 ++ .../state/BatchExecutionKeyedStateBackend.java | 5 + .../BatchExecutionInternalTimeServiceTest.java | 24 +-- ...utionInternalTimeServiceWithAsyncStateTest.java | 24 +-- .../state/BatchExecutionStateBackendTest.java | 162 ++++++++++----------- .../AbstractUdfStreamOperatorLifecycleTest.java | 1 + 10 files changed, 167 insertions(+), 105 deletions(-) diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java index 807bf83bc95..d40d619cd7d 100644 --- a/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java +++ b/flink-runtime/src/main/java/org/apache/flink/runtime/state/KeyedStateBackend.java @@ -60,6 +60,15 @@ public interface KeyedStateBackend<K> /** Act as a fast path for {@link #setCurrentKey} when the key group is known. */ void setCurrentKeyAndKeyGroup(K newKey, int newKeyGroupIndex); + /** + * Moves to the next key of a key-sorted input. Unlike {@link #setCurrentKey}, it must never go + * back to a key that was current before. Backends that do not depend on the key order treat it + * as {@link #setCurrentKey}. + */ + default void moveToNextKey(K newKey) { + setCurrentKey(newKey); + } + /** * @return Serializer of the key. */ diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java index d091128bdb5..1e2ed86ed8a 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperator.java @@ -615,6 +615,17 @@ public abstract class AbstractStreamOperator<OUT> stateHandler.setCurrentKey(key); } + /** + * Sets the key of the next record of a keyed input. Unlike {@link #setCurrentKey}, which an + * operator may call to switch to any key, it never goes back to a previous key of a key-sorted + * input. Override it, not {@link #setCurrentKey}, to act before the input moves on to the next + * key, e.g. to flush records buffered for the current key. + */ + @Override + public void moveToNextKey(Object key) { + stateHandler.moveToNextKey(key); + } + public Object getCurrentKey() { return stateHandler.getCurrentKey(); } diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java index 8dfd584351a..1b9e2e2df39 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/AbstractStreamOperatorV2.java @@ -503,6 +503,17 @@ public abstract class AbstractStreamOperatorV2<OUT> stateHandler.setCurrentKey(key); } + /** + * Sets the key of the next record of a keyed input. Unlike {@link #setCurrentKey}, which an + * operator may call to switch to any key, it never goes back to a previous key of a key-sorted + * input. Override it, not {@link #setCurrentKey}, to act before the input moves on to the next + * key, e.g. to flush records buffered for the current key. + */ + @Override + public void moveToNextKey(Object key) { + stateHandler.moveToNextKey(key); + } + @SuppressWarnings({"unchecked", "rawtypes"}) public Object getCurrentKey() { return stateHandler.getCurrentKey(); diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/KeyContext.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/KeyContext.java index 03c5d3d5645..3cd32112425 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/KeyContext.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/KeyContext.java @@ -28,5 +28,14 @@ public interface KeyContext { void setCurrentKey(Object key); + /** + * Moves to the next key of a key-sorted input, e.g. when the next record of a sorted input in + * BATCH execution belongs to a new key. Unlike {@link #setCurrentKey}, it must never go back to + * a key that was current before. In BATCH execution keys may only change this way. + */ + default void moveToNextKey(Object key) { + setCurrentKey(key); + } + Object getCurrentKey(); } diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java index 7a9b11b5b4e..a5c879c6578 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/StreamOperatorStateHandler.java @@ -464,6 +464,22 @@ public class StreamOperatorStateHandler { } } + /** See {@link KeyedStateBackend#moveToNextKey}. */ + @SuppressWarnings({"unchecked"}) + public void moveToNextKey(Object key) { + if (keyedStateBackend != null) { + try { + @SuppressWarnings("rawtypes") + CheckpointableKeyedStateBackend rawBackend = keyedStateBackend; + + rawBackend.moveToNextKey(key); + } catch (Exception e) { + throw new RuntimeException( + "Exception occurred while setting the current key context.", e); + } + } + } + public Object getCurrentKey() { if (keyedStateBackend != null) { return keyedStateBackend.getCurrentKey(); diff --git a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionKeyedStateBackend.java b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionKeyedStateBackend.java index 457dd413b9f..bfb127e0b9d 100644 --- a/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionKeyedStateBackend.java +++ b/flink-runtime/src/main/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionKeyedStateBackend.java @@ -106,6 +106,11 @@ public class BatchExecutionKeyedStateBackend<K> implements CheckpointableKeyedSt @Override public void setCurrentKey(K newKey) { + moveToNextKey(newKey); + } + + @Override + public void moveToNextKey(K newKey) { if (!Objects.equals(newKey, currentKey)) { notifyKeySelected(newKey); for (State value : states.values()) { diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceTest.java index f53eb41c7e9..a03f947a0f1 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceTest.java @@ -163,7 +163,7 @@ class BatchExecutionInternalTimeServiceTest { new VoidNamespaceSerializer(), LambdaTrigger.eventTimeTrigger(timer -> timers.add(timer.getTimestamp()))); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 123); // advancing the watermark should not fire timers @@ -172,7 +172,7 @@ class BatchExecutionInternalTimeServiceTest { timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 150); // changing the current key fires all timers - keyedStatedBackend.setCurrentKey(2); + keyedStatedBackend.moveToNextKey(2); assertThat(timers).containsExactly(150L); } @@ -202,9 +202,9 @@ class BatchExecutionInternalTimeServiceTest { new VoidNamespaceSerializer(), LambdaTrigger.eventTimeTrigger(timer -> timers.add(timer.getTimestamp()))); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 123); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); assertThat(timers).isEmpty(); } @@ -239,7 +239,7 @@ class BatchExecutionInternalTimeServiceTest { eventTimeTrigger.setTimerService(timerService); assertThat(timerService.currentWatermark()).isEqualTo(Long.MIN_VALUE); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 123); assertThat(timerService.currentWatermark()).isEqualTo(Long.MIN_VALUE); @@ -248,7 +248,7 @@ class BatchExecutionInternalTimeServiceTest { assertThat(timerService.currentWatermark()).isEqualTo(Long.MIN_VALUE); // changing the current key fires all timers - keyedStatedBackend.setCurrentKey(2); + keyedStatedBackend.moveToNextKey(2); assertThat(timerService.currentWatermark()).isEqualTo(Long.MIN_VALUE); timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 124); @@ -285,13 +285,13 @@ class BatchExecutionInternalTimeServiceTest { LambdaTrigger.processingTimeTrigger( timer -> timers.add(timer.getTimestamp()))); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); timerService.registerProcessingTimeTimer(VoidNamespace.INSTANCE, 150); // we should never register physical timers assertThat(processingTimeService.getNumActiveTimers()).isZero(); // changing the current key fires all timers - keyedStatedBackend.setCurrentKey(2); + keyedStatedBackend.moveToNextKey(2); assertThat(timers).containsExactly(150L); } @@ -327,13 +327,13 @@ class BatchExecutionInternalTimeServiceTest { "test", KEY_SERIALIZER, new VoidNamespaceSerializer(), trigger); trigger.setTimerService(timerService); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 150); // we should never register physical timers assertThat(processingTimeService.getNumActiveTimers()).isZero(); // changing the current key fires all timers - keyedStatedBackend.setCurrentKey(2); + keyedStatedBackend.moveToNextKey(2); // We check that the timer from the callback is ignored assertThat(timers).containsExactly(150L); @@ -370,13 +370,13 @@ class BatchExecutionInternalTimeServiceTest { "test", KEY_SERIALIZER, new VoidNamespaceSerializer(), trigger); trigger.setTimerService(timerService); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); timerService.registerProcessingTimeTimer(VoidNamespace.INSTANCE, 150); // we should never register physical timers assertThat(processingTimeService.getNumActiveTimers()).isZero(); // changing the current key fires all timers - keyedStatedBackend.setCurrentKey(2); + keyedStatedBackend.moveToNextKey(2); // We check that the timer from the callback is ignored assertThat(timers).containsExactly(150L); diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceWithAsyncStateTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceWithAsyncStateTest.java index cf2f24f3050..abb65aa090a 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceWithAsyncStateTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionInternalTimeServiceWithAsyncStateTest.java @@ -139,7 +139,7 @@ class BatchExecutionInternalTimeServiceWithAsyncStateTest { buildTimerService( LambdaTrigger.eventTimeTrigger(timer -> timers.add(timer.getTimestamp()))); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 123); // advancing the watermark should not fire timers @@ -148,7 +148,7 @@ class BatchExecutionInternalTimeServiceWithAsyncStateTest { timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 150); // changing the current key fires all timers - keyedStatedBackend.setCurrentKey(2); + keyedStatedBackend.moveToNextKey(2); assertThat(timers).containsExactly(150L); } @@ -160,9 +160,9 @@ class BatchExecutionInternalTimeServiceWithAsyncStateTest { buildTimerService( LambdaTrigger.eventTimeTrigger(timer -> timers.add(timer.getTimestamp()))); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 123); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); assertThat(timers).isEmpty(); } @@ -180,7 +180,7 @@ class BatchExecutionInternalTimeServiceWithAsyncStateTest { eventTimeTrigger.setTimerService(timerService); assertThat(timerService.currentWatermark()).isEqualTo(Long.MIN_VALUE); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 123); assertThat(timerService.currentWatermark()).isEqualTo(Long.MIN_VALUE); @@ -189,7 +189,7 @@ class BatchExecutionInternalTimeServiceWithAsyncStateTest { assertThat(timerService.currentWatermark()).isEqualTo(Long.MIN_VALUE); // changing the current key fires all timers - keyedStatedBackend.setCurrentKey(2); + keyedStatedBackend.moveToNextKey(2); assertThat(timerService.currentWatermark()).isEqualTo(Long.MIN_VALUE); timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 124); @@ -207,13 +207,13 @@ class BatchExecutionInternalTimeServiceWithAsyncStateTest { LambdaTrigger.processingTimeTrigger( timer -> timers.add(timer.getTimestamp()))); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); timerService.registerProcessingTimeTimer(VoidNamespace.INSTANCE, 150); // we should never register physical timers assertThat(processingTimeService.getNumActiveTimers()).isZero(); // changing the current key fires all timers - keyedStatedBackend.setCurrentKey(2); + keyedStatedBackend.moveToNextKey(2); assertThat(timers).containsExactly(150L); } @@ -231,13 +231,13 @@ class BatchExecutionInternalTimeServiceWithAsyncStateTest { InternalTimerService<VoidNamespace> timerService = buildTimerService(trigger); trigger.setTimerService(timerService); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); timerService.registerEventTimeTimer(VoidNamespace.INSTANCE, 150); // we should never register physical timers assertThat(processingTimeService.getNumActiveTimers()).isZero(); // changing the current key fires all timers - keyedStatedBackend.setCurrentKey(2); + keyedStatedBackend.moveToNextKey(2); // We check that the timer from the callback is ignored assertThat(timers).containsExactly(150L); @@ -256,13 +256,13 @@ class BatchExecutionInternalTimeServiceWithAsyncStateTest { InternalTimerService<VoidNamespace> timerService = buildTimerService(trigger); trigger.setTimerService(timerService); - keyedStatedBackend.setCurrentKey(1); + keyedStatedBackend.moveToNextKey(1); timerService.registerProcessingTimeTimer(VoidNamespace.INSTANCE, 150); // we should never register physical timers assertThat(processingTimeService.getNumActiveTimers()).isZero(); // changing the current key fires all timers - keyedStatedBackend.setCurrentKey(2); + keyedStatedBackend.moveToNextKey(2); // We check that the timer from the callback is ignored assertThat(timers).containsExactly(150L); diff --git a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackendTest.java b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackendTest.java index 70232ddfcda..ada5f796018 100644 --- a/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackendTest.java +++ b/flink-runtime/src/test/java/org/apache/flink/streaming/api/operators/sorted/state/BatchExecutionStateBackendTest.java @@ -91,7 +91,7 @@ class BatchExecutionStateBackendTest { keyedBackend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, stateDescr); - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); assertThat(state.get()).isNull(); assertThatThrownBy(() -> state.add(null)).isInstanceOf(NullPointerException.class); @@ -118,7 +118,7 @@ class BatchExecutionStateBackendTest { keyedBackend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, stateDescr); - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); assertThat(state.get()).isNull(); List<Long> adding = new ArrayList<>(); @@ -150,7 +150,7 @@ class BatchExecutionStateBackendTest { keyedBackend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, stateDescr); - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); assertThat(state.get()).isNull(); assertThatThrownBy(() -> state.addAll(null)).isInstanceOf(NullPointerException.class); @@ -177,7 +177,7 @@ class BatchExecutionStateBackendTest { keyedBackend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, stateDescr); - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); assertThat(state.get()).isNull(); List<Long> adding = new ArrayList<>(); @@ -209,7 +209,7 @@ class BatchExecutionStateBackendTest { keyedBackend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, stateDescr); - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); assertThat(state.get()).isNull(); assertThatThrownBy(() -> state.update(null)).isInstanceOf(NullPointerException.class); @@ -231,7 +231,7 @@ class BatchExecutionStateBackendTest { keyedBackend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, stateDescr); - keyedBackend.setCurrentKey("g"); + keyedBackend.moveToNextKey("g"); assertThat(state.get()).isNull(); assertThat(state.get()).isNull(); state.addAll(Collections.emptyList()); @@ -271,7 +271,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - abc spreads the values over three namespaces - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); state.setCurrentNamespace(namespace1); state.add(33L); state.add(55L); @@ -289,7 +289,7 @@ class BatchExecutionStateBackendTest { // make sure all lists / maps are cleared - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -315,7 +315,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - def spreads the values over two namespaces (one empty) - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); state.setCurrentNamespace(namespace1); state.add(11L); state.add(44L); @@ -325,14 +325,14 @@ class BatchExecutionStateBackendTest { state.add(55L); state.add(33L); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).containsExactlyInAnyOrder(11L, 22L, 33L, 44L, 55L); // make sure all lists / maps are cleared - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -358,14 +358,14 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - ghi is empty - keyedBackend.setCurrentKey("ghi"); + keyedBackend.moveToNextKey("ghi"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isNull(); // make sure all lists / maps are cleared - keyedBackend.setCurrentKey("ghi"); + keyedBackend.moveToNextKey("ghi"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -391,7 +391,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - jkl has all elements already in the target namespace - keyedBackend.setCurrentKey("jkl"); + keyedBackend.moveToNextKey("jkl"); state.setCurrentNamespace(namespace1); state.add(11L); state.add(22L); @@ -399,12 +399,12 @@ class BatchExecutionStateBackendTest { state.add(44L); state.add(55L); - keyedBackend.setCurrentKey("jkl"); + keyedBackend.moveToNextKey("jkl"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).containsExactlyInAnyOrder(11L, 22L, 33L, 44L, 55L); - keyedBackend.setCurrentKey("jkl"); + keyedBackend.moveToNextKey("jkl"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -430,7 +430,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - mno has all elements already in one source namespace - keyedBackend.setCurrentKey("mno"); + keyedBackend.moveToNextKey("mno"); state.setCurrentNamespace(namespace3); state.add(11L); state.add(22L); @@ -438,14 +438,14 @@ class BatchExecutionStateBackendTest { state.add(44L); state.add(55L); - keyedBackend.setCurrentKey("mno"); + keyedBackend.moveToNextKey("mno"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).containsExactlyInAnyOrder(11L, 22L, 33L, 44L, 55L); // make sure all lists / maps are cleared - keyedBackend.setCurrentKey("mno"); + keyedBackend.moveToNextKey("mno"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -464,28 +464,28 @@ class BatchExecutionStateBackendTest { keyedBackend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, stateDescr); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); assertThat(state.get()).isNull(); state.add(17L); state.add(11L); assertThat(state.get()).isEqualTo(28L); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); assertThat(state.get()).isEqualTo(28L); state.clear(); assertThat(state.get()).isNull(); - keyedBackend.setCurrentKey("g"); + keyedBackend.moveToNextKey("g"); assertThat(state.get()).isNull(); state.add(1L); state.add(2L); - keyedBackend.setCurrentKey("g"); + keyedBackend.moveToNextKey("g"); state.add(3L); state.add(2L); state.add(1L); - keyedBackend.setCurrentKey("g"); + keyedBackend.moveToNextKey("g"); assertThat(state.get()).isEqualTo(9L); } } @@ -511,7 +511,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - abc spreads the values over three namespaces - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); state.setCurrentNamespace(namespace1); state.add(33L); state.add(55L); @@ -523,12 +523,12 @@ class BatchExecutionStateBackendTest { state.setCurrentNamespace(namespace3); state.add(44L); - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isEqualTo(expectedResult); - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -556,7 +556,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - def spreads the values over two namespaces (one empty) - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); state.setCurrentNamespace(namespace1); state.add(11L); state.add(44L); @@ -566,12 +566,12 @@ class BatchExecutionStateBackendTest { state.add(55L); state.add(33L); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isEqualTo(expectedResult); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -597,7 +597,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - ghi is empty - keyedBackend.setCurrentKey("ghi"); + keyedBackend.moveToNextKey("ghi"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isNull(); @@ -625,7 +625,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - jkl has all elements already in the target namespace - keyedBackend.setCurrentKey("jkl"); + keyedBackend.moveToNextKey("jkl"); state.setCurrentNamespace(namespace1); state.add(11L); state.add(22L); @@ -633,12 +633,12 @@ class BatchExecutionStateBackendTest { state.add(44L); state.add(55L); - keyedBackend.setCurrentKey("jkl"); + keyedBackend.moveToNextKey("jkl"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isEqualTo(expectedResult); - keyedBackend.setCurrentKey("jkl"); + keyedBackend.moveToNextKey("jkl"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -666,7 +666,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - mno has all elements already in one source namespace - keyedBackend.setCurrentKey("mno"); + keyedBackend.moveToNextKey("mno"); state.setCurrentNamespace(namespace3); state.add(11L); state.add(22L); @@ -674,12 +674,12 @@ class BatchExecutionStateBackendTest { state.add(44L); state.add(55L); - keyedBackend.setCurrentKey("mno"); + keyedBackend.moveToNextKey("mno"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isEqualTo(expectedResult); - keyedBackend.setCurrentKey("mno"); + keyedBackend.moveToNextKey("mno"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -699,31 +699,31 @@ class BatchExecutionStateBackendTest { keyedBackend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, stateDescr); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); assertThat(state.get()).isNull(); state.add(17L); state.add(11L); assertThat(state.get()).isEqualTo(28L); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); assertThat(state.get()).isEqualTo(28L); state.clear(); assertThat(state.get()).isNull(); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); assertThat(state.get()).isNull(); - keyedBackend.setCurrentKey("g"); + keyedBackend.moveToNextKey("g"); assertThat(state.get()).isNull(); state.add(1L); state.add(2L); - keyedBackend.setCurrentKey("g"); + keyedBackend.moveToNextKey("g"); state.add(3L); state.add(2L); state.add(1L); - keyedBackend.setCurrentKey("g"); + keyedBackend.moveToNextKey("g"); assertThat(state.get()).isEqualTo(9L); state.clear(); assertThat(state.get()).isNull(); @@ -751,7 +751,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - abc spreads the values over three namespaces - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); state.setCurrentNamespace(namespace1); state.add(33L); state.add(55L); @@ -763,12 +763,12 @@ class BatchExecutionStateBackendTest { state.setCurrentNamespace(namespace3); state.add(44L); - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isEqualTo(expectedResult); - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -796,7 +796,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - def spreads the values over two namespaces (one empty) - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); state.setCurrentNamespace(namespace1); state.add(11L); state.add(44L); @@ -806,12 +806,12 @@ class BatchExecutionStateBackendTest { state.add(55L); state.add(33L); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isEqualTo(expectedResult); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -837,7 +837,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - ghi is empty - keyedBackend.setCurrentKey("ghi"); + keyedBackend.moveToNextKey("ghi"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isNull(); @@ -865,7 +865,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - jkl has all elements already in the target namespace - keyedBackend.setCurrentKey("jkl"); + keyedBackend.moveToNextKey("jkl"); state.setCurrentNamespace(namespace1); state.add(11L); state.add(22L); @@ -873,12 +873,12 @@ class BatchExecutionStateBackendTest { state.add(44L); state.add(55L); - keyedBackend.setCurrentKey("jkl"); + keyedBackend.moveToNextKey("jkl"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isEqualTo(expectedResult); - keyedBackend.setCurrentKey("jkl"); + keyedBackend.moveToNextKey("jkl"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -906,7 +906,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - mno has all elements already in one source namespace - keyedBackend.setCurrentKey("mno"); + keyedBackend.moveToNextKey("mno"); state.setCurrentNamespace(namespace3); state.add(11L); state.add(22L); @@ -914,12 +914,12 @@ class BatchExecutionStateBackendTest { state.add(44L); state.add(55L); - keyedBackend.setCurrentKey("mno"); + keyedBackend.moveToNextKey("mno"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isEqualTo(expectedResult); - keyedBackend.setCurrentKey("mno"); + keyedBackend.moveToNextKey("mno"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -939,28 +939,28 @@ class BatchExecutionStateBackendTest { keyedBackend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, stateDescr); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); assertThat(state.get()).isNull(); state.add(17L); state.add(11L); assertThat(state.get()).isEqualTo(28L); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); assertThat(state.get()).isEqualTo(28L); state.clear(); assertThat(state.get()).isNull(); - keyedBackend.setCurrentKey("g"); + keyedBackend.moveToNextKey("g"); assertThat(state.get()).isNull(); state.add(1L); state.add(2L); - keyedBackend.setCurrentKey("g"); + keyedBackend.moveToNextKey("g"); state.add(3L); state.add(2L); state.add(1L); - keyedBackend.setCurrentKey("g"); + keyedBackend.moveToNextKey("g"); assertThat(state.get()).isEqualTo(9L); state.clear(); assertThat(state.get()).isNull(); @@ -988,7 +988,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - abc spreads the values over three namespaces - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); state.setCurrentNamespace(namespace1); state.add(33L); state.add(55L); @@ -1000,12 +1000,12 @@ class BatchExecutionStateBackendTest { state.setCurrentNamespace(namespace3); state.add(44L); - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isEqualTo(expectedResult); - keyedBackend.setCurrentKey("abc"); + keyedBackend.moveToNextKey("abc"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -1033,7 +1033,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - def spreads the values over two namespaces (one empty) - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); state.setCurrentNamespace(namespace1); state.add(11L); state.add(44L); @@ -1043,12 +1043,12 @@ class BatchExecutionStateBackendTest { state.add(55L); state.add(33L); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isEqualTo(expectedResult); - keyedBackend.setCurrentKey("def"); + keyedBackend.moveToNextKey("def"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -1074,7 +1074,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - ghi is empty - keyedBackend.setCurrentKey("ghi"); + keyedBackend.moveToNextKey("ghi"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isNull(); @@ -1102,7 +1102,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - jkl has all elements already in the target namespace - keyedBackend.setCurrentKey("jkl"); + keyedBackend.moveToNextKey("jkl"); state.setCurrentNamespace(namespace1); state.add(11L); state.add(22L); @@ -1110,12 +1110,12 @@ class BatchExecutionStateBackendTest { state.add(44L); state.add(55L); - keyedBackend.setCurrentKey("jkl"); + keyedBackend.moveToNextKey("jkl"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isEqualTo(expectedResult); - keyedBackend.setCurrentKey("jkl"); + keyedBackend.moveToNextKey("jkl"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -1143,7 +1143,7 @@ class BatchExecutionStateBackendTest { // populate the different namespaces // - mno has all elements already in one source namespace - keyedBackend.setCurrentKey("mno"); + keyedBackend.moveToNextKey("mno"); state.setCurrentNamespace(namespace3); state.add(11L); state.add(22L); @@ -1151,12 +1151,12 @@ class BatchExecutionStateBackendTest { state.add(44L); state.add(55L); - keyedBackend.setCurrentKey("mno"); + keyedBackend.moveToNextKey("mno"); state.mergeNamespaces(namespace1, asList(namespace2, namespace3)); state.setCurrentNamespace(namespace1); assertThat(state.get()).isEqualTo(expectedResult); - keyedBackend.setCurrentKey("mno"); + keyedBackend.moveToNextKey("mno"); state.setCurrentNamespace(namespace1); state.clear(); assertThat(state.get()).isNull(); @@ -1175,7 +1175,7 @@ class BatchExecutionStateBackendTest { MapState<Integer, Long> state = backend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId); - backend.setCurrentKey(1); + backend.moveToNextKey(1); assertThat(state.isEmpty()).isTrue(); int stateSize = 1024; @@ -1211,7 +1211,7 @@ class BatchExecutionStateBackendTest { MapState<Integer, Long> state = backend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId); - backend.setCurrentKey(1); + backend.moveToNextKey(1); int stateSize = 4096; for (int i = 0; i < stateSize; i++) { state.put(i, i * 2L); @@ -1254,7 +1254,7 @@ class BatchExecutionStateBackendTest { backend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId); - backend.setCurrentKey(1); + backend.moveToNextKey(1); assertThat(state.value()).isNull(); state.update("Ciao"); @@ -1278,7 +1278,7 @@ class BatchExecutionStateBackendTest { backend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId); - backend.setCurrentKey(1); + backend.moveToNextKey(1); assertThat(state.value()).isEqualTo("Hello"); state.update("Ciao"); @@ -1303,7 +1303,7 @@ class BatchExecutionStateBackendTest { backend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId); - backend.setCurrentKey(1); + backend.moveToNextKey(1); assertThat(state.get()).isNull(); state.add("Ciao"); @@ -1327,7 +1327,7 @@ class BatchExecutionStateBackendTest { backend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId); - backend.setCurrentKey(1); + backend.moveToNextKey(1); assertThat(state.get()).isNull(); state.update(Arrays.asList("Ciao", "Bello")); @@ -1352,7 +1352,7 @@ class BatchExecutionStateBackendTest { backend.getPartitionedState( VoidNamespace.INSTANCE, VoidNamespaceSerializer.INSTANCE, kvId); - backend.setCurrentKey(1); + backend.moveToNextKey(1); assertThat(state.entries()).isNotNull(); assertThat(state.entries().iterator()).isExhausted(); diff --git a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java index fc85940d81e..912ad20e310 100644 --- a/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java +++ b/flink-streaming-java/src/test/java/org/apache/flink/streaming/api/operators/AbstractUdfStreamOperatorLifecycleTest.java @@ -102,6 +102,7 @@ class AbstractUdfStreamOperatorLifecycleTest { + "getOperatorAttributes[], " + "getOperatorID[], " + "initializeState[interface org.apache.flink.streaming.api.operators.StreamTaskStateInitializer], " + + "moveToNextKey[class java.lang.Object], " + "notifyCheckpointAborted[long], " + "notifyCheckpointComplete[long], " + "open[], "
