public JobSpecMonitor forConfig(Config localScopeConfig, MutableJobCatalog jobCatalog) throws IOException {
      Preconditions.checkArgument(localScopeConfig.hasPath(TOPIC_KEY));
      Config config = localScopeConfig.withFallback(DEFAULTS);

      String topic = config.getString(TOPIC_KEY);

      SchemaVersionWriter versionWriter;
      try {
        versionWriter = (SchemaVersionWriter) GobblinConstructorUtils.
            invokeLongestConstructor(Class.forName(config.getString(SCHEMA_VERSION_READER_CLASS)), config);
      } catch (ReflectiveOperationException roe) {
        throw new IllegalArgumentException(roe);
      }

      return new AvroJobSpecKafkaJobMonitor(topic, jobCatalog, config, versionWriter);
    }
public JobSpecMonitor forConfig ( Config localScopeConfig , MutableJobCatalog jobCatalog ) throws IOException { Preconditions . checkArgument ( localScopeConfig . hasPath ( TOPIC_KEY ) ) ; Config config = localScopeConfig . withFallback ( DEFAULTS ) ; String topic = config . getString ( TOPIC_KEY ) ; SchemaVersionWriter versionWriter ; try { versionWriter = ( SchemaVersionWriter ) GobblinConstructorUtils . invokeLongestConstructor ( Class . forName ( config . getString ( SCHEMA_VERSION_READER_CLASS ) ) , config ) ; } catch ( ReflectiveOperationException roe ) { throw new IllegalArgumentException ( roe ) ; } return new AvroJobSpecKafkaJobMonitor ( topic , jobCatalog , config , versionWriter ) ; }
Create a {@link AvroJobSpecKafkaJobMonitor} from an input {@link Config}. Useful for multiple monitors, where
the configuration of each monitor is scoped.
@param localScopeConfig The sub-{@link Config} for this monitor without any namespacing (e.g. the key for
topic should simply be "topic").
@throws IOException