ApplyStatefulOperator extends OneStreamStateHandlerOperator { @Inject private ApplyStatefulOperator( @Parameter(SerializedUdf.class) final String serializedObject, final ClassLoader classLoader) throws IOException, ClassNotFoundException { this(SerializeUtils.deserializeFromString(serializedObject, classLoader)); } @Inject private  ApplyStatefulOperator(
      @Parameter(SerializedUdf.class) final String serializedObject,
      final ClassLoader classLoader); @Inject  ApplyStatefulOperator(final ApplyStatefulFunction<IN, OUT> applyStatefulFunction); @Override void processLeftData(final MistDataEvent input); @Override void processLeftWatermark(final MistWatermarkEvent input); @Override Map<String, Object> getStateSnapshot(); @Override void setState(final Map<String, Object> loadedState); @Override void processLeftCheckpoint(final MistCheckpointEvent input);  }
@Test public void testApplyStatefulOperator() { final MistDataEvent data10 = new MistDataEvent(10, 0L); final MistDataEvent data20 = new MistDataEvent(20, 1L); final MistDataEvent data15 = new MistDataEvent(15, 2L); final MistDataEvent data30 = new MistDataEvent(30, 3L); final MistWatermarkEvent watermarkEvent = new MistWatermarkEvent(4L); final ApplyStatefulFunction applyStatefulFunction = new FindMaxIntFunction(); final ApplyStatefulOperator<Integer, Integer> applyStatefulOperator = new ApplyStatefulOperator<>(applyStatefulFunction); final List<MistEvent> result = new LinkedList<>(); applyStatefulOperator.setOutputEmitter(new OutputBufferEmitter(result)); applyStatefulOperator.processLeftData(data10); Assert.assertEquals(1, result.size()); Assert.assertEquals(data10, result.get(0)); applyStatefulOperator.processLeftData(data20); Assert.assertEquals(2, result.size()); Assert.assertEquals(data20, result.get(1)); applyStatefulOperator.processLeftData(data15); Assert.assertEquals(3, result.size()); Assert.assertTrue(result.get(2) instanceof MistDataEvent); Assert.assertEquals(20, ((MistDataEvent)result.get(2)).getValue()); Assert.assertEquals(2L, result.get(2).getTimestamp()); applyStatefulOperator.processLeftData(data30); Assert.assertEquals(4, result.size()); Assert.assertEquals(data30, result.get(3)); applyStatefulOperator.processLeftWatermark(watermarkEvent); Assert.assertEquals(5, result.size()); Assert.assertEquals(watermarkEvent, result.get(4)); }