@Test(expected = Exception.class) public void testInjectionExceptionOfOperator() throws IOException, ClassNotFoundException { final Map<String, String> conf = TextSocketSourceConfiguration.newBuilder() .setHostAddress("localhost") .setHostPort(13666) .build().getConfiguration(); generator.newOperator(conf, classLoader); }
@SuppressWarnings("unchecked") public Operator newOperator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.OperatorConf.OP_TYPE.name()); if (type.equals(ConfValues.OperatorType.MAP.name())) { return new MapOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.FILTER.name())) { return new FilterOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.FLAT_MAP.name())) { return new FlatMapOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.APPLY_STATEFUL.name())) { return new ApplyStatefulOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.STATE_TRANSITION.name())) { final String initialState = conf.get(ConfKeys.StateTransitionOperator.INITIAL_STATE.name()); final Map<String, Collection<Tuple2<MISTPredicate, String>>> stateTable = getObject(conf, ConfKeys.StateTransitionOperator.STATE_TABLE.name(), classLoader); final Set<String> finalState = getObject(conf, ConfKeys.StateTransitionOperator.FINAL_STATE.name(), classLoader); return new StateTransitionOperator(initialState, finalState, stateTable); } else if (type.equals(ConfValues.OperatorType.CEP.name())) { final List<CepEventPattern> cepEventPatterns = getObject(conf, ConfKeys.CepOperator.CEP_EVENT.name(), classLoader); final long windowTime = Long.valueOf(conf.get(ConfKeys.CepOperator.WINDOW_TIME.name())); return new CepOperator(cepEventPatterns, windowTime); } else if (type.equals(ConfValues.OperatorType.REDUCE_BY_KEY.name())) { final int keyFieldNum = Integer.valueOf(conf.get(ConfKeys.ReduceByKeyOperator.KEY_INDEX.name())); final MISTBiFunction reduceFunc = getObject(conf, ConfKeys.ReduceByKeyOperator.MIST_BI_FUNC.name(), classLoader); return new ReduceByKeyOperator(keyFieldNum, reduceFunc); } else if (type.equals(ConfValues.OperatorType.UNION.name())) { return new UnionOperator(); } else if (type.equals(ConfValues.OperatorType.TIME_WINDOW.name())) { final int windowSize = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_SIZE.name())); final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new TimeWindowOperator(windowSize, windowInterval); } else if (type.equals(ConfValues.OperatorType.COUNT_WINDOW.name())) { final int windowSize = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_SIZE.name())); final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new CountWindowOperator(windowSize, windowInterval); } else if (type.equals(ConfValues.OperatorType.SESSION_WINDOW.name())) { final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new SessionWindowOperator(windowInterval); } else if (type.equals(ConfValues.OperatorType.JOIN.name())) { return new JoinOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.AGGREGATE_WINDOW.name())) { return new AggregateWindowOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.APPLY_STATEFUL_WINDOW.name())) { return new ApplyStatefulWindowOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else { throw new RuntimeException("Invalid operator: " + type); } }
PhysicalObjectGenerator implements AutoCloseable { @SuppressWarnings("unchecked") public Operator newOperator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.OperatorConf.OP_TYPE.name()); if (type.equals(ConfValues.OperatorType.MAP.name())) { return new MapOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.FILTER.name())) { return new FilterOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.FLAT_MAP.name())) { return new FlatMapOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.APPLY_STATEFUL.name())) { return new ApplyStatefulOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.STATE_TRANSITION.name())) { final String initialState = conf.get(ConfKeys.StateTransitionOperator.INITIAL_STATE.name()); final Map<String, Collection<Tuple2<MISTPredicate, String>>> stateTable = getObject(conf, ConfKeys.StateTransitionOperator.STATE_TABLE.name(), classLoader); final Set<String> finalState = getObject(conf, ConfKeys.StateTransitionOperator.FINAL_STATE.name(), classLoader); return new StateTransitionOperator(initialState, finalState, stateTable); } else if (type.equals(ConfValues.OperatorType.CEP.name())) { final List<CepEventPattern> cepEventPatterns = getObject(conf, ConfKeys.CepOperator.CEP_EVENT.name(), classLoader); final long windowTime = Long.valueOf(conf.get(ConfKeys.CepOperator.WINDOW_TIME.name())); return new CepOperator(cepEventPatterns, windowTime); } else if (type.equals(ConfValues.OperatorType.REDUCE_BY_KEY.name())) { final int keyFieldNum = Integer.valueOf(conf.get(ConfKeys.ReduceByKeyOperator.KEY_INDEX.name())); final MISTBiFunction reduceFunc = getObject(conf, ConfKeys.ReduceByKeyOperator.MIST_BI_FUNC.name(), classLoader); return new ReduceByKeyOperator(keyFieldNum, reduceFunc); } else if (type.equals(ConfValues.OperatorType.UNION.name())) { return new UnionOperator(); } else if (type.equals(ConfValues.OperatorType.TIME_WINDOW.name())) { final int windowSize = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_SIZE.name())); final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new TimeWindowOperator(windowSize, windowInterval); } else if (type.equals(ConfValues.OperatorType.COUNT_WINDOW.name())) { final int windowSize = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_SIZE.name())); final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new CountWindowOperator(windowSize, windowInterval); } else if (type.equals(ConfValues.OperatorType.SESSION_WINDOW.name())) { final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new SessionWindowOperator(windowInterval); } else if (type.equals(ConfValues.OperatorType.JOIN.name())) { return new JoinOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.AGGREGATE_WINDOW.name())) { return new AggregateWindowOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.APPLY_STATEFUL_WINDOW.name())) { return new ApplyStatefulWindowOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else { throw new RuntimeException("Invalid operator: " + type); } } }
PhysicalObjectGenerator implements AutoCloseable { @SuppressWarnings("unchecked") public Operator newOperator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.OperatorConf.OP_TYPE.name()); if (type.equals(ConfValues.OperatorType.MAP.name())) { return new MapOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.FILTER.name())) { return new FilterOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.FLAT_MAP.name())) { return new FlatMapOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.APPLY_STATEFUL.name())) { return new ApplyStatefulOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.STATE_TRANSITION.name())) { final String initialState = conf.get(ConfKeys.StateTransitionOperator.INITIAL_STATE.name()); final Map<String, Collection<Tuple2<MISTPredicate, String>>> stateTable = getObject(conf, ConfKeys.StateTransitionOperator.STATE_TABLE.name(), classLoader); final Set<String> finalState = getObject(conf, ConfKeys.StateTransitionOperator.FINAL_STATE.name(), classLoader); return new StateTransitionOperator(initialState, finalState, stateTable); } else if (type.equals(ConfValues.OperatorType.CEP.name())) { final List<CepEventPattern> cepEventPatterns = getObject(conf, ConfKeys.CepOperator.CEP_EVENT.name(), classLoader); final long windowTime = Long.valueOf(conf.get(ConfKeys.CepOperator.WINDOW_TIME.name())); return new CepOperator(cepEventPatterns, windowTime); } else if (type.equals(ConfValues.OperatorType.REDUCE_BY_KEY.name())) { final int keyFieldNum = Integer.valueOf(conf.get(ConfKeys.ReduceByKeyOperator.KEY_INDEX.name())); final MISTBiFunction reduceFunc = getObject(conf, ConfKeys.ReduceByKeyOperator.MIST_BI_FUNC.name(), classLoader); return new ReduceByKeyOperator(keyFieldNum, reduceFunc); } else if (type.equals(ConfValues.OperatorType.UNION.name())) { return new UnionOperator(); } else if (type.equals(ConfValues.OperatorType.TIME_WINDOW.name())) { final int windowSize = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_SIZE.name())); final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new TimeWindowOperator(windowSize, windowInterval); } else if (type.equals(ConfValues.OperatorType.COUNT_WINDOW.name())) { final int windowSize = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_SIZE.name())); final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new CountWindowOperator(windowSize, windowInterval); } else if (type.equals(ConfValues.OperatorType.SESSION_WINDOW.name())) { final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new SessionWindowOperator(windowInterval); } else if (type.equals(ConfValues.OperatorType.JOIN.name())) { return new JoinOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.AGGREGATE_WINDOW.name())) { return new AggregateWindowOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.APPLY_STATEFUL_WINDOW.name())) { return new ApplyStatefulWindowOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else { throw new RuntimeException("Invalid operator: " + type); } } @Inject private  PhysicalObjectGenerator(final ScheduledExecutorServiceWrapper schedulerWrapper,
                                  final KafkaSharedResource kafkaSharedResource,
                                  final NettySharedResource nettySharedResource,
                                  final MQTTResource mqttSharedResource,
                                  @Parameter(PeriodicCheckpointPeriod.class) final long checkpointPeriod,
                                  final StringIdentifierFactory identifierFactory); }
PhysicalObjectGenerator implements AutoCloseable { @SuppressWarnings("unchecked") public Operator newOperator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.OperatorConf.OP_TYPE.name()); if (type.equals(ConfValues.OperatorType.MAP.name())) { return new MapOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.FILTER.name())) { return new FilterOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.FLAT_MAP.name())) { return new FlatMapOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.APPLY_STATEFUL.name())) { return new ApplyStatefulOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.STATE_TRANSITION.name())) { final String initialState = conf.get(ConfKeys.StateTransitionOperator.INITIAL_STATE.name()); final Map<String, Collection<Tuple2<MISTPredicate, String>>> stateTable = getObject(conf, ConfKeys.StateTransitionOperator.STATE_TABLE.name(), classLoader); final Set<String> finalState = getObject(conf, ConfKeys.StateTransitionOperator.FINAL_STATE.name(), classLoader); return new StateTransitionOperator(initialState, finalState, stateTable); } else if (type.equals(ConfValues.OperatorType.CEP.name())) { final List<CepEventPattern> cepEventPatterns = getObject(conf, ConfKeys.CepOperator.CEP_EVENT.name(), classLoader); final long windowTime = Long.valueOf(conf.get(ConfKeys.CepOperator.WINDOW_TIME.name())); return new CepOperator(cepEventPatterns, windowTime); } else if (type.equals(ConfValues.OperatorType.REDUCE_BY_KEY.name())) { final int keyFieldNum = Integer.valueOf(conf.get(ConfKeys.ReduceByKeyOperator.KEY_INDEX.name())); final MISTBiFunction reduceFunc = getObject(conf, ConfKeys.ReduceByKeyOperator.MIST_BI_FUNC.name(), classLoader); return new ReduceByKeyOperator(keyFieldNum, reduceFunc); } else if (type.equals(ConfValues.OperatorType.UNION.name())) { return new UnionOperator(); } else if (type.equals(ConfValues.OperatorType.TIME_WINDOW.name())) { final int windowSize = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_SIZE.name())); final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new TimeWindowOperator(windowSize, windowInterval); } else if (type.equals(ConfValues.OperatorType.COUNT_WINDOW.name())) { final int windowSize = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_SIZE.name())); final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new CountWindowOperator(windowSize, windowInterval); } else if (type.equals(ConfValues.OperatorType.SESSION_WINDOW.name())) { final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new SessionWindowOperator(windowInterval); } else if (type.equals(ConfValues.OperatorType.JOIN.name())) { return new JoinOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.AGGREGATE_WINDOW.name())) { return new AggregateWindowOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.APPLY_STATEFUL_WINDOW.name())) { return new ApplyStatefulWindowOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else { throw new RuntimeException("Invalid operator: " + type); } } @Inject private  PhysicalObjectGenerator(final ScheduledExecutorServiceWrapper schedulerWrapper,
                                  final KafkaSharedResource kafkaSharedResource,
                                  final NettySharedResource nettySharedResource,
                                  final MQTTResource mqttSharedResource,
                                  @Parameter(PeriodicCheckpointPeriod.class) final long checkpointPeriod,
                                  final StringIdentifierFactory identifierFactory); @SuppressWarnings("unchecked") EventGenerator<T> newEventGenerator(
      final Map<String, String> conf,
      final ClassLoader classLoader); @SuppressWarnings("unchecked") DataGenerator newDataGenerator(
      final Map<String, String> conf,
      final ClassLoader classLoader); @SuppressWarnings("unchecked") Operator newOperator(
      final Map<String, String> conf,
      final ClassLoader classLoader); @SuppressWarnings("unchecked") Sink<T> newSink(
      final Map<String, String> conf,
      final ClassLoader classLoader); @Override void close(); }
PhysicalObjectGenerator implements AutoCloseable { @SuppressWarnings("unchecked") public Operator newOperator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.OperatorConf.OP_TYPE.name()); if (type.equals(ConfValues.OperatorType.MAP.name())) { return new MapOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.FILTER.name())) { return new FilterOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.FLAT_MAP.name())) { return new FlatMapOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.APPLY_STATEFUL.name())) { return new ApplyStatefulOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.STATE_TRANSITION.name())) { final String initialState = conf.get(ConfKeys.StateTransitionOperator.INITIAL_STATE.name()); final Map<String, Collection<Tuple2<MISTPredicate, String>>> stateTable = getObject(conf, ConfKeys.StateTransitionOperator.STATE_TABLE.name(), classLoader); final Set<String> finalState = getObject(conf, ConfKeys.StateTransitionOperator.FINAL_STATE.name(), classLoader); return new StateTransitionOperator(initialState, finalState, stateTable); } else if (type.equals(ConfValues.OperatorType.CEP.name())) { final List<CepEventPattern> cepEventPatterns = getObject(conf, ConfKeys.CepOperator.CEP_EVENT.name(), classLoader); final long windowTime = Long.valueOf(conf.get(ConfKeys.CepOperator.WINDOW_TIME.name())); return new CepOperator(cepEventPatterns, windowTime); } else if (type.equals(ConfValues.OperatorType.REDUCE_BY_KEY.name())) { final int keyFieldNum = Integer.valueOf(conf.get(ConfKeys.ReduceByKeyOperator.KEY_INDEX.name())); final MISTBiFunction reduceFunc = getObject(conf, ConfKeys.ReduceByKeyOperator.MIST_BI_FUNC.name(), classLoader); return new ReduceByKeyOperator(keyFieldNum, reduceFunc); } else if (type.equals(ConfValues.OperatorType.UNION.name())) { return new UnionOperator(); } else if (type.equals(ConfValues.OperatorType.TIME_WINDOW.name())) { final int windowSize = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_SIZE.name())); final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new TimeWindowOperator(windowSize, windowInterval); } else if (type.equals(ConfValues.OperatorType.COUNT_WINDOW.name())) { final int windowSize = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_SIZE.name())); final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new CountWindowOperator(windowSize, windowInterval); } else if (type.equals(ConfValues.OperatorType.SESSION_WINDOW.name())) { final int windowInterval = Integer.valueOf(conf.get(ConfKeys.WindowOperator.WINDOW_INTERVAL.name())); return new SessionWindowOperator(windowInterval); } else if (type.equals(ConfValues.OperatorType.JOIN.name())) { return new JoinOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.AGGREGATE_WINDOW.name())) { return new AggregateWindowOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else if (type.equals(ConfValues.OperatorType.APPLY_STATEFUL_WINDOW.name())) { return new ApplyStatefulWindowOperator(getObject(conf, ConfKeys.OperatorConf.UDF_STRING.name(), classLoader)); } else { throw new RuntimeException("Invalid operator: " + type); } } @Inject private  PhysicalObjectGenerator(final ScheduledExecutorServiceWrapper schedulerWrapper,
                                  final KafkaSharedResource kafkaSharedResource,
                                  final NettySharedResource nettySharedResource,
                                  final MQTTResource mqttSharedResource,
                                  @Parameter(PeriodicCheckpointPeriod.class) final long checkpointPeriod,
                                  final StringIdentifierFactory identifierFactory); @SuppressWarnings("unchecked") EventGenerator<T> newEventGenerator(
      final Map<String, String> conf,
      final ClassLoader classLoader); @SuppressWarnings("unchecked") DataGenerator newDataGenerator(
      final Map<String, String> conf,
      final ClassLoader classLoader); @SuppressWarnings("unchecked") Operator newOperator(
      final Map<String, String> conf,
      final ClassLoader classLoader); @SuppressWarnings("unchecked") Sink<T> newSink(
      final Map<String, String> conf,
      final ClassLoader classLoader); @Override void close();  }