[gobblin] branch master updated: [GOBBLIN-1534] Allow kafka metrics/events reporting to be disabled individually (#3386)

[ap...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

aplex pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/gobblin.git


The following commit(s) were added to refs/heads/master by this push:
     new bea2817  [GOBBLIN-1534] Allow kafka metrics/events reporting to be disabled individually (#3386)
bea2817 is described below

commit bea281712ad86f31634d5f6441fc492639cdf346
Author: Jack Moseley <jm...@linkedin.com>
AuthorDate: Tue Sep 7 11:14:15 2021 -0700

    [GOBBLIN-1534] Allow kafka metrics/events reporting to be disabled individually (#3386)
    
    Currently in both gaas and jobs launched by gaas, we use kafka event reporting, but not kafka metrics reporting. In #3035, there were some changes which includes splitting the kafka reporter factory. As a side effect, the two reporters (metrics/events) became all or nothing, where previously if just one was missing config it would be skipped. This causes gaas and jobs launched by gaas to always contain an error stacktrace about failing to instantiate the kafka metric reporter.
    
    This PR adds config keys to enable/disable the reporters individually. For backwards compatibility, if metrics.reporting.kafka.enabled is true, they will both be enabled.
---
 .../gobblin/configuration/ConfigurationKeys.java   |  4 ++++
 .../org/apache/gobblin/metrics/GobblinMetrics.java | 24 ++++++++++++++--------
 2 files changed, 19 insertions(+), 9 deletions(-)

diff --git a/gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java b/gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java
index 2c9b80c..feb3f92 100644
--- a/gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java
+++ b/gobblin-api/src/main/java/org/apache/gobblin/configuration/ConfigurationKeys.java
@@ -752,6 +752,10 @@ public class ConfigurationKeys {
   public static final String METRICS_REPORTING_KAFKA_ENABLED_KEY =
       METRICS_CONFIGURATIONS_PREFIX + "reporting.kafka.enabled";
   public static final String DEFAULT_METRICS_REPORTING_KAFKA_ENABLED = Boolean.toString(false);
+  public static final String METRICS_REPORTING_KAFKA_METRICS_ENABLED_KEY =
+      METRICS_CONFIGURATIONS_PREFIX + "reporting.kafka.metrics.enabled";
+  public static final String METRICS_REPORTING_KAFKA_EVENTS_ENABLED_KEY =
+      METRICS_CONFIGURATIONS_PREFIX + "reporting.kafka.events.enabled";
   public static final String DEFAULT_METRICS_REPORTING_KAFKA_REPORTER_CLASS =
       "org.apache.gobblin.metrics.kafka.KafkaMetricReporterFactory";
   public static final String DEFAULT_EVENTS_REPORTING_KAFKA_REPORTER_CLASS =
diff --git a/gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/GobblinMetrics.java b/gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/GobblinMetrics.java
index 4e4fa6c..9441d4f 100644
--- a/gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/GobblinMetrics.java
+++ b/gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/GobblinMetrics.java
@@ -612,20 +612,26 @@ public class GobblinMetrics {
   private void buildKafkaMetricReporter(Properties properties)
       throws MultiReporterException {
     List<MetricReporterException> reporterExceptions = Lists.newArrayList();
-    if (!Boolean.valueOf(properties.getProperty(ConfigurationKeys.METRICS_REPORTING_KAFKA_ENABLED_KEY,
+    if (!Boolean.parseBoolean(properties.getProperty(ConfigurationKeys.METRICS_REPORTING_KAFKA_ENABLED_KEY,
         ConfigurationKeys.DEFAULT_METRICS_REPORTING_KAFKA_ENABLED))) {
       return;
     }
 
-    try {
-      buildScheduledReporter(properties, ConfigurationKeys.DEFAULT_METRICS_REPORTING_KAFKA_REPORTER_CLASS);
-    } catch (MetricReporterException e) {
-      reporterExceptions.add(e);
+    if (Boolean.parseBoolean(properties.getProperty(ConfigurationKeys.METRICS_REPORTING_KAFKA_METRICS_ENABLED_KEY,
+        Boolean.toString(true)))) {
+      try {
+        buildScheduledReporter(properties, ConfigurationKeys.DEFAULT_METRICS_REPORTING_KAFKA_REPORTER_CLASS);
+      } catch (MetricReporterException e) {
+        reporterExceptions.add(e);
+      }
     }
-    try {
-      buildScheduledReporter(properties, ConfigurationKeys.DEFAULT_EVENTS_REPORTING_KAFKA_REPORTER_CLASS);
-    } catch (MetricReporterException e) {
-      reporterExceptions.add(e);
+    if (Boolean.parseBoolean(properties.getProperty(ConfigurationKeys.METRICS_REPORTING_KAFKA_EVENTS_ENABLED_KEY,
+        Boolean.toString(true)))) {
+      try {
+        buildScheduledReporter(properties, ConfigurationKeys.DEFAULT_EVENTS_REPORTING_KAFKA_REPORTER_CLASS);
+      } catch (MetricReporterException e) {
+        reporterExceptions.add(e);
+      }
     }
     if (!reporterExceptions.isEmpty()) {
       throw new MultiReporterException("Failed to start one or more Kafka reporters", reporterExceptions);