@Test public void testAddMetricReportingDynamicConfig() throws IOException { KafkaAvroSchemaRegistry schemaRegistry = Mockito.mock(KafkaAvroSchemaRegistry.class); Mockito.when(schemaRegistry.register(Mockito.any(Schema.class), Mockito.anyString())).thenAnswer(new Answer<String>() { @Override public String answer(InvocationOnMock invocation) { return "testId"; } }); Config config = ConfigFactory.empty().withValue(ConfigurationKeys.METRICS_KAFKA_TOPIC_EVENTS, ConfigValueFactory.fromAnyRef("topic")) .withValue(ConfigurationKeys.METRICS_REPORTING_KAFKA_ENABLED_KEY, ConfigValueFactory.fromAnyRef(true)) .withValue(ConfigurationKeys.METRICS_REPORTING_KAFKA_USE_SCHEMA_REGISTRY, ConfigValueFactory.fromAnyRef(true)) .withValue(KafkaSchemaRegistry.KAFKA_SCHEMA_REGISTRY_URL, ConfigValueFactory.fromAnyRef("http: config = GobblinYarnAppLauncher.addMetricReportingDynamicConfig(config, schemaRegistry); Assert.assertEquals(config.getString(ConfigurationKeys.METRICS_REPORTING_EVENTS_KAFKA_AVRO_SCHEMA_ID), "testId"); Assert.assertFalse(config.hasPath(ConfigurationKeys.METRICS_REPORTING_METRICS_KAFKA_AVRO_SCHEMA_ID)); }
@VisibleForTesting static Config addMetricReportingDynamicConfig(Config config, KafkaAvroSchemaRegistry registry) throws IOException { Properties properties = ConfigUtils.configToProperties(config); if (KafkaReporterUtils.isEventsEnabled(properties)) { Schema schema = KafkaReporterUtils.getGobblinTrackingEventSchema(); String schemaId = registry.register(schema, KafkaReporterUtils.getEventsTopic(properties).get()); LOGGER.info("Adding schemaId {} for GobblinTrackingEvent to the config", schemaId); config = config.withValue(ConfigurationKeys.METRICS_REPORTING_EVENTS_KAFKA_AVRO_SCHEMA_ID, ConfigValueFactory.fromAnyRef(schemaId)); } if (KafkaReporterUtils.isMetricsEnabled(properties)) { Schema schema = KafkaReporterUtils.getMetricReportSchema(); String schemaId = registry.register(schema, KafkaReporterUtils.getMetricsTopic(properties).get()); LOGGER.info("Adding schemaId {} for MetricReport to the config", schemaId); config = config.withValue(ConfigurationKeys.METRICS_REPORTING_METRICS_KAFKA_AVRO_SCHEMA_ID, ConfigValueFactory.fromAnyRef(schemaId)); } return config; }
GobblinYarnAppLauncher { @VisibleForTesting static Config addMetricReportingDynamicConfig(Config config, KafkaAvroSchemaRegistry registry) throws IOException { Properties properties = ConfigUtils.configToProperties(config); if (KafkaReporterUtils.isEventsEnabled(properties)) { Schema schema = KafkaReporterUtils.getGobblinTrackingEventSchema(); String schemaId = registry.register(schema, KafkaReporterUtils.getEventsTopic(properties).get()); LOGGER.info("Adding schemaId {} for GobblinTrackingEvent to the config", schemaId); config = config.withValue(ConfigurationKeys.METRICS_REPORTING_EVENTS_KAFKA_AVRO_SCHEMA_ID, ConfigValueFactory.fromAnyRef(schemaId)); } if (KafkaReporterUtils.isMetricsEnabled(properties)) { Schema schema = KafkaReporterUtils.getMetricReportSchema(); String schemaId = registry.register(schema, KafkaReporterUtils.getMetricsTopic(properties).get()); LOGGER.info("Adding schemaId {} for MetricReport to the config", schemaId); config = config.withValue(ConfigurationKeys.METRICS_REPORTING_METRICS_KAFKA_AVRO_SCHEMA_ID, ConfigValueFactory.fromAnyRef(schemaId)); } return config; } }
GobblinYarnAppLauncher { @VisibleForTesting static Config addMetricReportingDynamicConfig(Config config, KafkaAvroSchemaRegistry registry) throws IOException { Properties properties = ConfigUtils.configToProperties(config); if (KafkaReporterUtils.isEventsEnabled(properties)) { Schema schema = KafkaReporterUtils.getGobblinTrackingEventSchema(); String schemaId = registry.register(schema, KafkaReporterUtils.getEventsTopic(properties).get()); LOGGER.info("Adding schemaId {} for GobblinTrackingEvent to the config", schemaId); config = config.withValue(ConfigurationKeys.METRICS_REPORTING_EVENTS_KAFKA_AVRO_SCHEMA_ID, ConfigValueFactory.fromAnyRef(schemaId)); } if (KafkaReporterUtils.isMetricsEnabled(properties)) { Schema schema = KafkaReporterUtils.getMetricReportSchema(); String schemaId = registry.register(schema, KafkaReporterUtils.getMetricsTopic(properties).get()); LOGGER.info("Adding schemaId {} for MetricReport to the config", schemaId); config = config.withValue(ConfigurationKeys.METRICS_REPORTING_METRICS_KAFKA_AVRO_SCHEMA_ID, ConfigValueFactory.fromAnyRef(schemaId)); } return config; } GobblinYarnAppLauncher(Config config, YarnConfiguration yarnConfiguration); }
GobblinYarnAppLauncher { @VisibleForTesting static Config addMetricReportingDynamicConfig(Config config, KafkaAvroSchemaRegistry registry) throws IOException { Properties properties = ConfigUtils.configToProperties(config); if (KafkaReporterUtils.isEventsEnabled(properties)) { Schema schema = KafkaReporterUtils.getGobblinTrackingEventSchema(); String schemaId = registry.register(schema, KafkaReporterUtils.getEventsTopic(properties).get()); LOGGER.info("Adding schemaId {} for GobblinTrackingEvent to the config", schemaId); config = config.withValue(ConfigurationKeys.METRICS_REPORTING_EVENTS_KAFKA_AVRO_SCHEMA_ID, ConfigValueFactory.fromAnyRef(schemaId)); } if (KafkaReporterUtils.isMetricsEnabled(properties)) { Schema schema = KafkaReporterUtils.getMetricReportSchema(); String schemaId = registry.register(schema, KafkaReporterUtils.getMetricsTopic(properties).get()); LOGGER.info("Adding schemaId {} for MetricReport to the config", schemaId); config = config.withValue(ConfigurationKeys.METRICS_REPORTING_METRICS_KAFKA_AVRO_SCHEMA_ID, ConfigValueFactory.fromAnyRef(schemaId)); } return config; } GobblinYarnAppLauncher(Config config, YarnConfiguration yarnConfiguration); void launch(); synchronized void stop(); @Subscribe void handleApplicationReportArrivalEvent(ApplicationReportArrivalEvent applicationReportArrivalEvent); @Subscribe void handleGetApplicationReportFailureEvent(
      GetApplicationReportFailureEvent getApplicationReportFailureEvent); static void main(String[] args); }
GobblinYarnAppLauncher { @VisibleForTesting static Config addMetricReportingDynamicConfig(Config config, KafkaAvroSchemaRegistry registry) throws IOException { Properties properties = ConfigUtils.configToProperties(config); if (KafkaReporterUtils.isEventsEnabled(properties)) { Schema schema = KafkaReporterUtils.getGobblinTrackingEventSchema(); String schemaId = registry.register(schema, KafkaReporterUtils.getEventsTopic(properties).get()); LOGGER.info("Adding schemaId {} for GobblinTrackingEvent to the config", schemaId); config = config.withValue(ConfigurationKeys.METRICS_REPORTING_EVENTS_KAFKA_AVRO_SCHEMA_ID, ConfigValueFactory.fromAnyRef(schemaId)); } if (KafkaReporterUtils.isMetricsEnabled(properties)) { Schema schema = KafkaReporterUtils.getMetricReportSchema(); String schemaId = registry.register(schema, KafkaReporterUtils.getMetricsTopic(properties).get()); LOGGER.info("Adding schemaId {} for MetricReport to the config", schemaId); config = config.withValue(ConfigurationKeys.METRICS_REPORTING_METRICS_KAFKA_AVRO_SCHEMA_ID, ConfigValueFactory.fromAnyRef(schemaId)); } return config; } GobblinYarnAppLauncher(Config config, YarnConfiguration yarnConfiguration); void launch(); synchronized void stop(); @Subscribe void handleApplicationReportArrivalEvent(ApplicationReportArrivalEvent applicationReportArrivalEvent); @Subscribe void handleGetApplicationReportFailureEvent(
      GetApplicationReportFailureEvent getApplicationReportFailureEvent); static void main(String[] args); static final String GOBBLIN_YARN_CONFIG_OUTPUT_PATH; static final String GOBBLIN_YARN_APP_LAUNCHER_MODE; static final String DEFAULT_GOBBLIN_YARN_APP_LAUNCHER_MODE; static final String AZKABAN_APP_LAUNCHER_MODE_KEY; }