@Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp = "Bad config passed to TestTriggerer") public void testWithoutRightTriggererConfig() throws Exception { Map<String, Object> badConfig = new HashMap<>(); badConfig.put("something", "else"); testBatchConfig.setDiscoveryTriggererConfig(badConfig); config.put(BatchSourceConfig.BATCHSOURCE_CONFIG_KEY, new Gson().toJson(testBatchConfig)); batchSourceExecutor.open(config, context); }
@Override public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception { this.config = config; this.sourceContext = sourceContext; this.intermediateTopicName = SourceConfigUtils.computeBatchSourceIntermediateTopicName(sourceContext.getTenant(), sourceContext.getNamespace(), sourceContext.getSourceName()).toString(); this.getBatchSourceConfigs(config); this.initializeBatchSource(); this.start(); }
BatchSourceExecutor implements Source<T> { @Override public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception { this.config = config; this.sourceContext = sourceContext; this.intermediateTopicName = SourceConfigUtils.computeBatchSourceIntermediateTopicName(sourceContext.getTenant(), sourceContext.getNamespace(), sourceContext.getSourceName()).toString(); this.getBatchSourceConfigs(config); this.initializeBatchSource(); this.start(); } }
BatchSourceExecutor implements Source<T> { @Override public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception { this.config = config; this.sourceContext = sourceContext; this.intermediateTopicName = SourceConfigUtils.computeBatchSourceIntermediateTopicName(sourceContext.getTenant(), sourceContext.getNamespace(), sourceContext.getSourceName()).toString(); this.getBatchSourceConfigs(config); this.initializeBatchSource(); this.start(); }  }
BatchSourceExecutor implements Source<T> { @Override public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception { this.config = config; this.sourceContext = sourceContext; this.intermediateTopicName = SourceConfigUtils.computeBatchSourceIntermediateTopicName(sourceContext.getTenant(), sourceContext.getNamespace(), sourceContext.getSourceName()).toString(); this.getBatchSourceConfigs(config); this.initializeBatchSource(); this.start(); }  @Override void open(Map<String, Object> config, SourceContext sourceContext); @Override Record<T> read(); @Override void close(); }
BatchSourceExecutor implements Source<T> { @Override public void open(Map<String, Object> config, SourceContext sourceContext) throws Exception { this.config = config; this.sourceContext = sourceContext; this.intermediateTopicName = SourceConfigUtils.computeBatchSourceIntermediateTopicName(sourceContext.getTenant(), sourceContext.getNamespace(), sourceContext.getSourceName()).toString(); this.getBatchSourceConfigs(config); this.initializeBatchSource(); this.start(); }  @Override void open(Map<String, Object> config, SourceContext sourceContext); @Override Record<T> read(); @Override void close();  }