code: 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);
      }
docs: 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