ExecutableStageDoFnOperator extends DoFnOperator<InputT, OutputT> { @Override public void setCurrentKey(Object key) {} ExecutableStageDoFnOperator(
      String stepName,
      Coder<WindowedValue<InputT>> windowedInputCoder,
      Map<TupleTag<?>, Coder<?>> outputCoders,
      TupleTag<OutputT> mainOutputTag,
      List<TupleTag<?>> additionalOutputTags,
      OutputManagerFactory<OutputT> outputManagerFactory,
      Map<Integer, PCollectionView<?>> sideInputTagMapping,
      Collection<PCollectionView<?>> sideInputs,
      Map<RunnerApi.ExecutableStagePayload.SideInputId, PCollectionView<?>> sideInputIds,
      PipelineOptions options,
      RunnerApi.ExecutableStagePayload payload,
      JobInfo jobInfo,
      FlinkExecutableStageContextFactory contextFactory,
      Map<String, TupleTag<?>> outputMap,
      WindowingStrategy windowingStrategy,
      Coder keyCoder,
      KeySelector<WindowedValue<InputT>, ?> keySelector); @Override void open(); @Override void setKeyContextElement1(StreamRecord record); @Override void setCurrentKey(Object key); @Override ByteBuffer getCurrentKey(); @Override void close(); @Override void dispose(); @Override long applyInputWatermarkHold(long inputWatermark); @Override long applyOutputWatermarkHold(long currentOutputWatermark, long potentialOutputWatermark);  }
@Test public void testEnsureStateCleanupWithKeyedInputCleanupTimer() { InMemoryTimerInternals inMemoryTimerInternals = new InMemoryTimerInternals(); KeyedStateBackend keyedStateBackend = Mockito.mock(KeyedStateBackend.class); Lock stateBackendLock = Mockito.mock(Lock.class); StringUtf8Coder keyCoder = StringUtf8Coder.of(); IntervalWindow window = new IntervalWindow(new Instant(0), new Instant(10)); Coder<IntervalWindow> windowCoder = IntervalWindow.getCoder(); ExecutableStageDoFnOperator.CleanupTimer cleanupTimer = new ExecutableStageDoFnOperator.CleanupTimer<>( inMemoryTimerInternals, stateBackendLock, WindowingStrategy.globalDefault(), keyCoder, windowCoder, keyedStateBackend); cleanupTimer.setForWindow(KV.of("key", "string"), window); Mockito.verify(stateBackendLock).lock(); ByteBuffer key = FlinkKeyUtils.encodeKey("key", keyCoder); Mockito.verify(keyedStateBackend).setCurrentKey(key); assertThat( inMemoryTimerInternals.getNextTimer(TimeDomain.EVENT_TIME), is(window.maxTimestamp().plus(1))); Mockito.verify(stateBackendLock).unlock(); }