@Test @SuppressWarnings("unchecked") public void testReduceByKeyOperatorGetState() throws InterruptedException { final List<MistDataEvent> inputStream = ImmutableList.of( createTupleEvent("a", 1, 1L), createTupleEvent("b", 1, 2L), createTupleEvent("c", 1, 3L), createTupleEvent("a", 1, 4L), createTupleEvent("d", 1, 5L), createTupleEvent("a", 1, 6L), createTupleEvent("b", 1, 7L)); final Map<String, Integer> expectedOperatorState = new HashMap<>(); expectedOperatorState.put("a", 3); expectedOperatorState.put("b", 2); expectedOperatorState.put("c", 1); expectedOperatorState.put("d", 1); final int keyIndex = 0; final MISTBiFunction<Integer, Integer, Integer> wordCountFunc = (oldVal, val) -> oldVal + val; final ReduceByKeyOperator<String, Integer> wcOperator = new ReduceByKeyOperator<>(keyIndex, wordCountFunc); final List<MistEvent> result = new LinkedList<>(); wcOperator.setOutputEmitter(new OutputBufferEmitter(result)); inputStream.stream().forEach(wcOperator::processLeftData); final Map<String, Object> operatorStateMap = wcOperator.getStateSnapshot(); final Map<String, Integer> operatorState = (Map<String, Integer>)operatorStateMap.get("reduceByKeyState"); Assert.assertEquals(expectedOperatorState, operatorState); }
@Override public Map<String, Object> getStateSnapshot() { final Map<String, Object> stateMap = new HashMap<>(); stateMap.put("reduceByKeyState", new Cloner(new ObjenesisInstantiationStrategy()).deepClone(state)); return stateMap; }
ReduceByKeyOperator extends OneStreamStateHandlerOperator { @Override public Map<String, Object> getStateSnapshot() { final Map<String, Object> stateMap = new HashMap<>(); stateMap.put("reduceByKeyState", new Cloner(new ObjenesisInstantiationStrategy()).deepClone(state)); return stateMap; } }
ReduceByKeyOperator extends OneStreamStateHandlerOperator { @Override public Map<String, Object> getStateSnapshot() { final Map<String, Object> stateMap = new HashMap<>(); stateMap.put("reduceByKeyState", new Cloner(new ObjenesisInstantiationStrategy()).deepClone(state)); return stateMap; } @Inject private  ReduceByKeyOperator(
      @Parameter(KeyIndex.class) final int keyIndex,
      @Parameter(SerializedUdf.class) final String serializedObject,
      final ClassLoader classLoader); @Inject  ReduceByKeyOperator(@Parameter(KeyIndex.class) final int keyIndex,
                             final MISTBiFunction<V, V, V> reduceFunc); }
ReduceByKeyOperator extends OneStreamStateHandlerOperator { @Override public Map<String, Object> getStateSnapshot() { final Map<String, Object> stateMap = new HashMap<>(); stateMap.put("reduceByKeyState", new Cloner(new ObjenesisInstantiationStrategy()).deepClone(state)); return stateMap; } @Inject private  ReduceByKeyOperator(
      @Parameter(KeyIndex.class) final int keyIndex,
      @Parameter(SerializedUdf.class) final String serializedObject,
      final ClassLoader classLoader); @Inject  ReduceByKeyOperator(@Parameter(KeyIndex.class) final int keyIndex,
                             final MISTBiFunction<V, V, V> reduceFunc); @Override void processLeftData(final MistDataEvent input); @Override void processLeftWatermark(final MistWatermarkEvent input); @Override Map<String, Object> getStateSnapshot(); @SuppressWarnings("unchecked") @Override void setState(final Map<String, Object> loadedState); @Override void processLeftCheckpoint(final MistCheckpointEvent input); }
ReduceByKeyOperator extends OneStreamStateHandlerOperator { @Override public Map<String, Object> getStateSnapshot() { final Map<String, Object> stateMap = new HashMap<>(); stateMap.put("reduceByKeyState", new Cloner(new ObjenesisInstantiationStrategy()).deepClone(state)); return stateMap; } @Inject private  ReduceByKeyOperator(
      @Parameter(KeyIndex.class) final int keyIndex,
      @Parameter(SerializedUdf.class) final String serializedObject,
      final ClassLoader classLoader); @Inject  ReduceByKeyOperator(@Parameter(KeyIndex.class) final int keyIndex,
                             final MISTBiFunction<V, V, V> reduceFunc); @Override void processLeftData(final MistDataEvent input); @Override void processLeftWatermark(final MistWatermarkEvent input); @Override Map<String, Object> getStateSnapshot(); @SuppressWarnings("unchecked") @Override void setState(final Map<String, Object> loadedState); @Override void processLeftCheckpoint(final MistCheckpointEvent input);  }