[GOBBLIN-2239] Introduce pushMessagesSync for sync GTE emission (#4156)

* introduce pushMessagesSync for GTE emission

* add missing java doc

* some refactoring

* updated java doc

Author: Blazer-007

gobblin-metrics-libs/gobblin-metrics-base/src/main/java/org/apache/gobblin/metrics/reporter/EventReporter.java

@@ -163,12 +163,29 @@ public void report() {
     reportEventQueue(this.reportingQueue);
   }
 
+  /**
+   * Report all {@link org.apache.gobblin.metrics.GobblinTrackingEvent}s in the queue synchronously.
+   */
+  public void reportSynchronously() {
+    reportEventQueueSynchronously(this.reportingQueue);
+  }
+
   /**
    * Emit all {@link org.apache.gobblin.metrics.GobblinTrackingEvent} in queue.
    * @param queue {@link java.util.Queue} containing {@link org.apache.gobblin.metrics.GobblinTrackingEvent}s that should be emitted.
    */
   public abstract void reportEventQueue(Queue<GobblinTrackingEvent> queue);
 
+  /**
+   * Emit all {@link org.apache.gobblin.metrics.GobblinTrackingEvent} in queue synchronously.
+   * Default implementation provides no synchronous guarantees and just calls {@link #reportEventQueue(Queue)}.
+   * Subclasses should override this method to implement true synchronous reporting behavior.
+   * @param queue {@link java.util.Queue} containing {@link org.apache.gobblin.metrics.GobblinTrackingEvent}s that should be emitted.
+   */
+  public void reportEventQueueSynchronously(Queue<GobblinTrackingEvent> queue) {
+    reportEventQueue(queue);
+  }
+
   /**
    * NOOP because {@link com.codahale.metrics.ScheduledReporter} requires this method implemented.
    */

gobblin-metrics-libs/gobblin-metrics/src/main/java/org/apache/gobblin/metrics/GobblinMetrics.java

@@ -350,6 +350,18 @@ public Timer getTimer(String prefix, String... suffixes) {
     return this.metricContext.timer(MetricRegistry.name(prefix, suffixes));
   }
 
+  /**
+   * Get the first {@link ScheduledReporter} of the given class if it exists.
+   * @param clazz the class of the ScheduledReporter to get
+   * @return an Optional containing the first ScheduledReporter of the given class if it exists, else an absent Optional
+   */
+  public <T extends com.codahale.metrics.ScheduledReporter> java.util.Optional<T> getScheduledReporter(Class<T> clazz) {
+    if (clazz == null) {
+      throw new IllegalArgumentException("Class argument cannot be null");
+    }
+    return this.codahaleScheduledReporters.stream().filter(clazz::isInstance).map(clazz::cast).findFirst();
+  }
+
   /**
    * Starts metric reporting and appends the given metrics file suffix to the current value of
    * {@link ConfigurationKeys#METRICS_FILE_SUFFIX}.

gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/KafkaEventKeyValueReporter.java

@@ -54,6 +54,29 @@ protected KafkaEventKeyValueReporter(Builder<?> builder) throws IOException {
 
   @Override
   public void reportEventQueue(Queue<GobblinTrackingEvent> queue) {
+    reportEventQueueInternal(queue, false);
+  }
+
+  @Override
+  public void reportEventQueueSynchronously(Queue<GobblinTrackingEvent> queue) {
+    reportEventQueueInternal(queue, true);
+  }
+
+  private void reportEventQueueInternal(Queue<GobblinTrackingEvent> queue, boolean sync) {
+    List<Pair<String, byte[]>> events = getEventsFromQueue(queue);
+    if (!events.isEmpty()) {
+      log.info("Pushing {} Gobblin Tracking Events to Kafka", events.size());
+      if (sync) {
+        this.kafkaPusher.pushMessagesSync(events);
+      } else {
+        this.kafkaPusher.pushMessages(events);
+      }
+    } else {
+      log.debug("No GTE to push.");
+    }
+  }
+
+  private List<Pair<String, byte[]>> getEventsFromQueue(Queue<GobblinTrackingEvent> queue) {
     GobblinTrackingEvent nextEvent;
     List<Pair<String, byte[]>> events = Lists.newArrayList();
 
@@ -75,9 +98,7 @@ public void reportEventQueue(Queue<GobblinTrackingEvent> queue) {
       events.add(Pair.of(key, this.serializer.serializeRecord(nextEvent)));
     }
 
-    if (!events.isEmpty()) {
-      this.kafkaPusher.pushMessages(events);
-    }
+    return events;
   }
 
   private static class BuilderImpl extends Builder<BuilderImpl> {

gobblin-modules/gobblin-kafka-common/src/main/java/org/apache/gobblin/metrics/kafka/Pusher.java

@@ -27,7 +27,16 @@
 public interface Pusher<M> extends Closeable {
   /**
    * Push all byte array messages to the Kafka topic.
-   * @param messages List of byte array messages to push to Kakfa.
+   * @param messages List of byte array messages to push to Kafka.
    */
   void pushMessages(List<M> messages);
+
+  /**
+   * Synchronous version of {@link #pushMessages(List)}.
+   * Default implementation just calls {@link #pushMessages(List)}.
+   * @param messages List of byte array messages to push to Kafka.
+   */
+  default void pushMessagesSync(List<M> messages) {
+    pushMessages(messages);
+  }
 }

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/GobblinTemporalConfigurationKeys.java

@@ -46,6 +46,10 @@ public interface GobblinTemporalConfigurationKeys {
   String GOBBLIN_TEMPORAL_WORK_DIR_CLEANUP_ENABLED = PREFIX + "work.dir.cleanup.enabled";
   String DEFAULT_GOBBLIN_TEMPORAL_WORK_DIR_CLEANUP_ENABLED = "true";
 
+  String GOBBLIN_TEMPORAL_CONTAINER_METRICS_PREFIX = PREFIX + "container.metrics.";
+  String GOBBLIN_TEMPORAL_CONTAINER_METRICS_APPLICATION_NAME = GOBBLIN_TEMPORAL_CONTAINER_METRICS_PREFIX + "application.name";
+  String GOBBLIN_TEMPORAL_CONTAINER_METRICS_TASK_RUNNER_ID = GOBBLIN_TEMPORAL_CONTAINER_METRICS_PREFIX + "task.runner.id";
+
   /**
    * Suffix for metrics emitted by GobblinTemporalJobLauncher for preventing collisions with prod jobs
    * during testing

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/cluster/GobblinTemporalTaskRunner.java

@@ -208,8 +208,10 @@ private Path initAppWorkDir(Config config, Optional<Path> appWorkDirOptional) {
 
   private Config saveConfigToFile(Config config)
       throws IOException {
-    Config newConf = config
-        .withValue(CLUSTER_APP_WORK_DIR, ConfigValueFactory.fromAnyRef(this.appWorkPath.toString()));
+    // Save the application name and task runner id in the config for getting ContainerMetrics instance later
+    Config newConf = config.withValue(CLUSTER_APP_WORK_DIR, ConfigValueFactory.fromAnyRef(this.appWorkPath.toString()))
+        .withValue(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_APPLICATION_NAME, ConfigValueFactory.fromAnyRef(this.applicationName))
+        .withValue(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_TASK_RUNNER_ID, ConfigValueFactory.fromAnyRef(this.taskRunnerId));
     ConfigUtils configUtils = new ConfigUtils(new FileUtils());
     configUtils.saveConfigToFile(newConf, CLUSTER_CONF_PATH);
     return newConf;

gobblin-temporal/src/main/java/org/apache/gobblin/temporal/workflows/metrics/SubmitGTEActivityImpl.java

@@ -19,22 +19,50 @@
 
 import java.util.Map;
 
-import io.temporal.workflow.Workflow;
-
-import org.slf4j.Logger;
+import java.util.Optional;
+import lombok.extern.slf4j.Slf4j;
+import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
 
+import org.apache.gobblin.cluster.ContainerMetrics;
 import org.apache.gobblin.metrics.event.EventSubmitter;
 import org.apache.gobblin.metrics.event.GobblinEventBuilder;
 import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.metrics.kafka.KafkaAvroEventKeyValueReporter;
+import org.apache.gobblin.temporal.GobblinTemporalConfigurationKeys;
+import org.apache.gobblin.temporal.cluster.WorkerConfig;
+import org.apache.gobblin.util.ConfigUtils;
 
 
+@Slf4j
 public class SubmitGTEActivityImpl implements SubmitGTEActivity {
-    private static Logger log = Workflow.getLogger(SubmitGTEActivityImpl.class);
 
     @Override
     public void submitGTE(GobblinEventBuilder eventBuilder, EventSubmitterContext eventSubmitterContext) {
-        log.info("submitting GTE - {}", summarizeEventMetadataForLogging(eventBuilder));
+        log.info("Submitting GTE - {}", summarizeEventMetadataForLogging(eventBuilder));
         eventSubmitterContext.create().submit(eventBuilder);
+        Config config = WorkerConfig.of(this).orElse(ConfigFactory.load());
+        if (config.hasPath(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_TASK_RUNNER_ID) &&
+            config.hasPath(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_APPLICATION_NAME)) {
+            String containerMetricsApplicationName = config.getString(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_APPLICATION_NAME);
+            String containerMetricsTaskRunnerId = config.getString(GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_TASK_RUNNER_ID);
+            ContainerMetrics containerMetrics = ContainerMetrics.get(ConfigUtils.configToState(config), containerMetricsApplicationName, containerMetricsTaskRunnerId);
+            log.info("Fetched container metrics instance {} for taskRunnerId: {} and applicationName: {}", containerMetrics.toString(),
+                containerMetricsTaskRunnerId, containerMetricsApplicationName);
+            Optional<KafkaAvroEventKeyValueReporter> kafkaReporterOptional = containerMetrics.getScheduledReporter(KafkaAvroEventKeyValueReporter.class);
+            if (kafkaReporterOptional.isPresent()) {
+                log.info("Submitting GTE in synchronous manner to Kafka reporter");
+                kafkaReporterOptional.get().reportSynchronously();
+                log.info("Submitted GTE to Kafka reporter");
+            } else {
+                log.warn("No KafkaAvroEventKeyValueReporter found in container metrics for taskRunnerId: {} and applicationName: {}",
+                    containerMetricsTaskRunnerId, containerMetricsApplicationName);
+            }
+        } else {
+            log.warn("Both {} and {} should be set to fetch container metrics instance for synchronous GTE submission",
+                GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_TASK_RUNNER_ID,
+                GobblinTemporalConfigurationKeys.GOBBLIN_TEMPORAL_CONTAINER_METRICS_APPLICATION_NAME);
+        }
     }
 
     private static String summarizeEventMetadataForLogging(GobblinEventBuilder eventBuilder) {