@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.discoveryThread = Executors.newSingleThreadExecutor(
      new DefaultThreadFactory(
        String.format("%s-batch-source-discovery",
          FunctionCommon.getFullyQualifiedName(
            sourceContext.getTenant(), sourceContext.getNamespace(), sourceContext.getSourceName()))));
    this.getBatchSourceConfigs(config);
    this.initializeBatchSource();
    this.start();
  }
@Test(expectedExceptions = IllegalArgumentException.class, expectedExceptionsMessageRegExp =
          "Bad config passed to TestBatchPushSource")
  public void testPushWithoutRightSourceConfig() throws Exception {
    pushConfig.remove("foo");
    pushConfig.put("something", "else");
    batchSourceExecutor.open(pushConfig, context);
  }