@Test(expected = Exception.class) public void testInjectionExceptionOfEventGenerator() throws IOException, ClassNotFoundException { generator.newEventGenerator(new HashMap<>(), classLoader); }
@SuppressWarnings("unchecked") public <T> EventGenerator<T> newEventGenerator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.Watermark.EVENT_GENERATOR.name()); final String tefString = conf.get(ConfKeys.SourceConf.TIMESTAMP_EXTRACT_FUNC.name()); final MISTFunction timestampExtractFunc; if (tefString == null) { timestampExtractFunc = null; } else { timestampExtractFunc = SerializeUtils.deserializeFromString( conf.get(ConfKeys.SourceConf.TIMESTAMP_EXTRACT_FUNC.name()), classLoader); } if (type.equals(ConfValues.EventGeneratorType.PERIODIC_EVENT_GEN.name())) { final long period = Long.valueOf(conf.get(ConfKeys.Watermark.PERIODIC_WATERMARK_PERIOD.name())); final long delay = Long.valueOf(conf.get(ConfKeys.Watermark.PERIODIC_WATERMARK_DELAY.name())); return new PeriodicEventGenerator( timestampExtractFunc, period, checkpointPeriod, delay, watermarkTimeUnit, scheduler); } else if (type.equals(ConfValues.EventGeneratorType.PUNCTUATED_EVENT_GEN.name())) { final MISTPredicate watermarkPredicate = SerializeUtils.deserializeFromString( conf.get(ConfKeys.Watermark.WATERMARK_PREDICATE.name()), classLoader); final WatermarkTimestampFunction tf = SerializeUtils.deserializeFromString( conf.get(ConfKeys.Watermark.TIMESTAMP_PARSE_OBJECT.name()), classLoader); return new PunctuatedEventGenerator( timestampExtractFunc, watermarkPredicate, tf, checkpointPeriod, watermarkTimeUnit, scheduler); } else { throw new RuntimeException("Invalid event generator: " + type); } }
PhysicalObjectGenerator implements AutoCloseable { @SuppressWarnings("unchecked") public <T> EventGenerator<T> newEventGenerator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.Watermark.EVENT_GENERATOR.name()); final String tefString = conf.get(ConfKeys.SourceConf.TIMESTAMP_EXTRACT_FUNC.name()); final MISTFunction timestampExtractFunc; if (tefString == null) { timestampExtractFunc = null; } else { timestampExtractFunc = SerializeUtils.deserializeFromString( conf.get(ConfKeys.SourceConf.TIMESTAMP_EXTRACT_FUNC.name()), classLoader); } if (type.equals(ConfValues.EventGeneratorType.PERIODIC_EVENT_GEN.name())) { final long period = Long.valueOf(conf.get(ConfKeys.Watermark.PERIODIC_WATERMARK_PERIOD.name())); final long delay = Long.valueOf(conf.get(ConfKeys.Watermark.PERIODIC_WATERMARK_DELAY.name())); return new PeriodicEventGenerator( timestampExtractFunc, period, checkpointPeriod, delay, watermarkTimeUnit, scheduler); } else if (type.equals(ConfValues.EventGeneratorType.PUNCTUATED_EVENT_GEN.name())) { final MISTPredicate watermarkPredicate = SerializeUtils.deserializeFromString( conf.get(ConfKeys.Watermark.WATERMARK_PREDICATE.name()), classLoader); final WatermarkTimestampFunction tf = SerializeUtils.deserializeFromString( conf.get(ConfKeys.Watermark.TIMESTAMP_PARSE_OBJECT.name()), classLoader); return new PunctuatedEventGenerator( timestampExtractFunc, watermarkPredicate, tf, checkpointPeriod, watermarkTimeUnit, scheduler); } else { throw new RuntimeException("Invalid event generator: " + type); } } }
PhysicalObjectGenerator implements AutoCloseable { @SuppressWarnings("unchecked") public <T> EventGenerator<T> newEventGenerator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.Watermark.EVENT_GENERATOR.name()); final String tefString = conf.get(ConfKeys.SourceConf.TIMESTAMP_EXTRACT_FUNC.name()); final MISTFunction timestampExtractFunc; if (tefString == null) { timestampExtractFunc = null; } else { timestampExtractFunc = SerializeUtils.deserializeFromString( conf.get(ConfKeys.SourceConf.TIMESTAMP_EXTRACT_FUNC.name()), classLoader); } if (type.equals(ConfValues.EventGeneratorType.PERIODIC_EVENT_GEN.name())) { final long period = Long.valueOf(conf.get(ConfKeys.Watermark.PERIODIC_WATERMARK_PERIOD.name())); final long delay = Long.valueOf(conf.get(ConfKeys.Watermark.PERIODIC_WATERMARK_DELAY.name())); return new PeriodicEventGenerator( timestampExtractFunc, period, checkpointPeriod, delay, watermarkTimeUnit, scheduler); } else if (type.equals(ConfValues.EventGeneratorType.PUNCTUATED_EVENT_GEN.name())) { final MISTPredicate watermarkPredicate = SerializeUtils.deserializeFromString( conf.get(ConfKeys.Watermark.WATERMARK_PREDICATE.name()), classLoader); final WatermarkTimestampFunction tf = SerializeUtils.deserializeFromString( conf.get(ConfKeys.Watermark.TIMESTAMP_PARSE_OBJECT.name()), classLoader); return new PunctuatedEventGenerator( timestampExtractFunc, watermarkPredicate, tf, checkpointPeriod, watermarkTimeUnit, scheduler); } else { throw new RuntimeException("Invalid event generator: " + 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 <T> EventGenerator<T> newEventGenerator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.Watermark.EVENT_GENERATOR.name()); final String tefString = conf.get(ConfKeys.SourceConf.TIMESTAMP_EXTRACT_FUNC.name()); final MISTFunction timestampExtractFunc; if (tefString == null) { timestampExtractFunc = null; } else { timestampExtractFunc = SerializeUtils.deserializeFromString( conf.get(ConfKeys.SourceConf.TIMESTAMP_EXTRACT_FUNC.name()), classLoader); } if (type.equals(ConfValues.EventGeneratorType.PERIODIC_EVENT_GEN.name())) { final long period = Long.valueOf(conf.get(ConfKeys.Watermark.PERIODIC_WATERMARK_PERIOD.name())); final long delay = Long.valueOf(conf.get(ConfKeys.Watermark.PERIODIC_WATERMARK_DELAY.name())); return new PeriodicEventGenerator( timestampExtractFunc, period, checkpointPeriod, delay, watermarkTimeUnit, scheduler); } else if (type.equals(ConfValues.EventGeneratorType.PUNCTUATED_EVENT_GEN.name())) { final MISTPredicate watermarkPredicate = SerializeUtils.deserializeFromString( conf.get(ConfKeys.Watermark.WATERMARK_PREDICATE.name()), classLoader); final WatermarkTimestampFunction tf = SerializeUtils.deserializeFromString( conf.get(ConfKeys.Watermark.TIMESTAMP_PARSE_OBJECT.name()), classLoader); return new PunctuatedEventGenerator( timestampExtractFunc, watermarkPredicate, tf, checkpointPeriod, watermarkTimeUnit, scheduler); } else { throw new RuntimeException("Invalid event generator: " + 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 <T> EventGenerator<T> newEventGenerator( final Map<String, String> conf, final ClassLoader classLoader) throws IOException, ClassNotFoundException { final String type = conf.get(ConfKeys.Watermark.EVENT_GENERATOR.name()); final String tefString = conf.get(ConfKeys.SourceConf.TIMESTAMP_EXTRACT_FUNC.name()); final MISTFunction timestampExtractFunc; if (tefString == null) { timestampExtractFunc = null; } else { timestampExtractFunc = SerializeUtils.deserializeFromString( conf.get(ConfKeys.SourceConf.TIMESTAMP_EXTRACT_FUNC.name()), classLoader); } if (type.equals(ConfValues.EventGeneratorType.PERIODIC_EVENT_GEN.name())) { final long period = Long.valueOf(conf.get(ConfKeys.Watermark.PERIODIC_WATERMARK_PERIOD.name())); final long delay = Long.valueOf(conf.get(ConfKeys.Watermark.PERIODIC_WATERMARK_DELAY.name())); return new PeriodicEventGenerator( timestampExtractFunc, period, checkpointPeriod, delay, watermarkTimeUnit, scheduler); } else if (type.equals(ConfValues.EventGeneratorType.PUNCTUATED_EVENT_GEN.name())) { final MISTPredicate watermarkPredicate = SerializeUtils.deserializeFromString( conf.get(ConfKeys.Watermark.WATERMARK_PREDICATE.name()), classLoader); final WatermarkTimestampFunction tf = SerializeUtils.deserializeFromString( conf.get(ConfKeys.Watermark.TIMESTAMP_PARSE_OBJECT.name()), classLoader); return new PunctuatedEventGenerator( timestampExtractFunc, watermarkPredicate, tf, checkpointPeriod, watermarkTimeUnit, scheduler); } else { throw new RuntimeException("Invalid event generator: " + 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();  }