feat: capture AM bootstrap failures in GaaS issues[] array

When the Gobblin YARN ApplicationMaster fails to start (e.g. token
refresh failures, HDFS quota exceeded), the Gobblin job runtime never
launches, so no Kafka events are emitted and GaaS sees the job stuck in
ORCHESTRATED forever.

Fix: AzkabanGobblinYarnAppLauncher now initialises a ServiceBasedAppLauncher
before launching the YARN app, following the same pattern as
IvyJobLauncherTemporal. On launch() failure it emits a JOB_FAILED timing
event (to transition GaaS job status to FAILED) and an IssueEventBuilder
(to populate the issues[] array with the root-cause stack trace).

The ServiceBasedAppLauncher correctly resolves the SSL keystore path via
XinfraDynamicConfigGenerator reading from the Hadoop token file, fixing
the "./var/identity.p12 not found" error in GGW pods. Its stop() in the
finally block flushes all events to Kafka via Xinfra before process exit.

Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

Author: agam-99

gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/azkaban/AzkabanGobblinLocalYarnAppLauncher.java

@@ -29,8 +29,7 @@
  * capability of changing yarn-resource related configuration in the way that could work with lighter hardware.
  */
 public class AzkabanGobblinLocalYarnAppLauncher extends AzkabanGobblinYarnAppLauncher {
-  public AzkabanGobblinLocalYarnAppLauncher(String jobId, Properties gobblinProps)
-      throws IOException {
+  public AzkabanGobblinLocalYarnAppLauncher(String jobId, Properties gobblinProps) throws Exception {
     super(jobId, gobblinProps);
   }
 

gobblin-modules/gobblin-azkaban/src/main/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncher.java

@@ -18,20 +18,43 @@
 package org.apache.gobblin.azkaban;
 
 import java.io.IOException;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
+import java.util.UUID;
 import java.util.concurrent.TimeoutException;
 
+import org.apache.commons.codec.digest.DigestUtils;
+import org.apache.commons.lang3.exception.ExceptionUtils;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.log4j.Level;
 import org.apache.log4j.Logger;
 
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Maps;
 import com.typesafe.config.Config;
+import com.typesafe.config.ConfigFactory;
+import com.typesafe.config.ConfigValue;
 import com.typesafe.config.ConfigValueFactory;
 
 import azkaban.jobExecutor.AbstractJob;
 import lombok.Getter;
 
+import org.apache.gobblin.configuration.ConfigurationKeys;
+import org.apache.gobblin.configuration.DynamicConfigGenerator;
+import org.apache.gobblin.instrumented.Instrumented;
+import org.apache.gobblin.metrics.MetricContext;
+import org.apache.gobblin.metrics.event.EventSubmitter;
+import org.apache.gobblin.metrics.event.TimingEvent;
+import org.apache.gobblin.runtime.DynamicConfigGeneratorFactory;
+import org.apache.gobblin.runtime.app.ApplicationLauncher;
+import org.apache.gobblin.runtime.app.ServiceBasedAppLauncher;
+import org.apache.gobblin.runtime.troubleshooter.Issue;
+import org.apache.gobblin.runtime.troubleshooter.IssueEventBuilder;
+import org.apache.gobblin.runtime.troubleshooter.IssueSeverity;
 import org.apache.gobblin.util.AzkabanLauncherUtils;
 import org.apache.gobblin.util.ConfigUtils;
 import org.apache.gobblin.yarn.GobblinYarnAppLauncher;
@@ -56,29 +79,89 @@
 public class AzkabanGobblinYarnAppLauncher extends AbstractJob {
   private static final Logger LOGGER = Logger.getLogger(AzkabanGobblinYarnAppLauncher.class);
 
+  /** Config prefix for the {@link ServiceBasedAppLauncher} used for bootstrap-failure event reporting. */
+  private static final String APP_LAUNCHER_PREFIX = "gobblinYarnAppLauncher";
+
   private final GobblinYarnAppLauncher gobblinYarnAppLauncher;
+  private final Properties gobblinProps;
+  private final ApplicationLauncher applicationLauncher;
+  private final EventSubmitter eventSubmitter;
 
   @Getter
   protected final YarnConfiguration yarnConfiguration;
 
-  public AzkabanGobblinYarnAppLauncher(String jobId, Properties gobblinProps)
-      throws IOException {
+  public AzkabanGobblinYarnAppLauncher(String jobId, Properties gobblinProps) throws Exception {
     super(jobId, LOGGER);
     gobblinProps = AzkabanLauncherUtils.undoPlaceholderConversion(gobblinProps);
     addRuntimeProperties(gobblinProps);
+    this.gobblinProps = gobblinProps;
 
-    Config gobblinConfig = ConfigUtils.propertiesToConfig(gobblinProps);
+    // Populate correct SSL keystore path from the Hadoop token file so Xinfra reporters work in
+    // the GGW pod (XinfraDynamicConfigGenerator writes the cert to a temp file and sets the path)
+    Config propsAsConfig = ConfigUtils.propertiesToConfig(gobblinProps);
+    DynamicConfigGenerator dynamicConfigGenerator =
+        DynamicConfigGeneratorFactory.createDynamicConfigGenerator(propsAsConfig);
+    Config dynamicConfig = dynamicConfigGenerator.generateDynamicConfig(propsAsConfig);
+    for (Map.Entry<String, ConfigValue> entry : dynamicConfig.entrySet()) {
+      gobblinProps.put(entry.getKey(), entry.getValue().unwrapped().toString());
+    }
 
-    //Suppress logs from classes that emit Yarn application Id that Azkaban uses to kill the application.
-    setLogLevelForClasses(gobblinConfig);
+    this.applicationLauncher = buildApplicationLauncher(gobblinProps);
+    this.eventSubmitter = buildEventSubmitter();
 
+    Config gobblinConfig = ConfigUtils.propertiesToConfig(gobblinProps);
+    setLogLevelForClasses(gobblinConfig);
     yarnConfiguration = initYarnConf(gobblinProps);
-
     gobblinConfig = gobblinConfig.withValue(GobblinYarnAppLauncher.GOBBLIN_YARN_APP_LAUNCHER_MODE,
         ConfigValueFactory.fromAnyRef(GobblinYarnAppLauncher.AZKABAN_APP_LAUNCHER_MODE_KEY));
     this.gobblinYarnAppLauncher = getYarnAppLauncher(gobblinConfig);
   }
 
+  @VisibleForTesting
+  AzkabanGobblinYarnAppLauncher(String jobId, Properties gobblinProps,
+      GobblinYarnAppLauncher gobblinYarnAppLauncher, ApplicationLauncher applicationLauncher,
+      EventSubmitter eventSubmitter) throws IOException {
+    super(jobId, LOGGER);
+    this.gobblinProps = gobblinProps;
+    this.gobblinYarnAppLauncher = gobblinYarnAppLauncher;
+    this.applicationLauncher = applicationLauncher;
+    this.eventSubmitter = eventSubmitter;
+    this.yarnConfiguration = new YarnConfiguration();
+  }
+
+  /**
+   * Creates and starts a {@link ServiceBasedAppLauncher} scoped for bootstrap-failure event reporting.
+   * Custom metric reporters that require YARN-container-specific SSL setup are temporarily removed
+   * so the launcher starts cleanly in GGW pods, then restored for the actual Gobblin job launcher.
+   */
+  private ApplicationLauncher buildApplicationLauncher(Properties props) throws Exception {
+    String customBuilders = props.containsKey(ConfigurationKeys.METRICS_CUSTOM_BUILDERS)
+        ? (String) props.remove(ConfigurationKeys.METRICS_CUSTOM_BUILDERS) : null;
+    try {
+      Config fullConfig = ConfigUtils.propertiesToConfig(props);
+      Config appConfig =
+          ConfigUtils.getConfigOrEmpty(fullConfig, APP_LAUNCHER_PREFIX).withFallback(fullConfig);
+      ApplicationLauncher launcher = new ServiceBasedAppLauncher(
+          ConfigUtils.configToProperties(appConfig), "GobblinYarnAppLauncher-" + UUID.randomUUID());
+      launcher.start();
+      return launcher;
+    } finally {
+      if (customBuilders != null) {
+        props.put(ConfigurationKeys.METRICS_CUSTOM_BUILDERS, customBuilders);
+      }
+    }
+  }
+
+  /**
+   * Builds an {@link EventSubmitter} from the {@link MetricContext} wired up by
+   * {@link #buildApplicationLauncher}, following the pattern used by {@code IvyJobLauncherTemporal}.
+   */
+  private EventSubmitter buildEventSubmitter() {
+    MetricContext metricContext =
+        Instrumented.getMetricContext(ConfigUtils.configToState(ConfigFactory.empty()), getClass());
+    return new EventSubmitter.Builder(metricContext, "org.apache.gobblin.service").build();
+  }
+
   protected GobblinYarnAppLauncher getYarnAppLauncher(Config gobblinConfig)
       throws IOException {
     GobblinYarnAppLauncher gobblinYarnAppLauncher = new GobblinYarnAppLauncher(gobblinConfig, this.yarnConfiguration);
@@ -116,7 +199,16 @@ protected void addRuntimeProperties(Properties gobblinProps) {
 
   @Override
   public void run() throws Exception {
-    this.gobblinYarnAppLauncher.launch();
+    try {
+      this.gobblinYarnAppLauncher.launch();
+    } catch (Exception e) {
+      LOGGER.error("Failed to launch the Gobblin Yarn application", e);
+      submitJobFailedEvent(e);
+      throw e;
+    } finally {
+      // Triggers MetricsReportingService shutdown which flushes events to Kafka before process exits.
+      this.applicationLauncher.stop();
+    }
 
     Runtime.getRuntime().addShutdownHook(new Thread() {
 
@@ -138,4 +230,51 @@ public void run() {
   public void cancel() throws Exception {
     this.gobblinYarnAppLauncher.stop();
   }
+
+  /**
+   * Emits a JOB_FAILED timing event and an {@link IssueEventBuilder} so GaaS transitions the job
+   * to FAILED and populates {@code issues[]} with the bootstrap failure root cause.
+   */
+  private void submitJobFailedEvent(Exception e) {
+    this.eventSubmitter.getTimingEvent(TimingEvent.LauncherTimings.JOB_FAILED).stop(getFlowMetadata());
+
+    // Emit an IssueEventBuilder so the error details appear in the issues[] array in GaaS
+    String summary = "YARN AM bootstrap failed: " + ExceptionUtils.getRootCauseMessage(e);
+    Issue issue = Issue.builder()
+        .time(ZonedDateTime.now(ZoneOffset.UTC))
+        .severity(IssueSeverity.ERROR)
+        .code("T" + DigestUtils.sha256Hex(summary).substring(0, 6).toUpperCase())
+        .summary(summary)
+        .details(ExceptionUtils.getStackTrace(e))
+        .sourceClass(AzkabanGobblinYarnAppLauncher.class.getName())
+        .properties(Collections.emptyMap())
+        .build();
+    IssueEventBuilder issueEventBuilder = new IssueEventBuilder(IssueEventBuilder.JOB_ISSUE);
+    issueEventBuilder.setIssue(issue);
+    issueEventBuilder.addMetadata(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD,
+        this.gobblinProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY, ""));
+    issueEventBuilder.addMetadata(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD,
+        this.gobblinProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY, ""));
+    issueEventBuilder.addMetadata(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+        this.gobblinProps.getProperty(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, ""));
+    issueEventBuilder.addMetadata(TimingEvent.FlowEventConstants.JOB_NAME_FIELD,
+        this.gobblinProps.getProperty(ConfigurationKeys.JOB_NAME_KEY, ""));
+    this.eventSubmitter.submit(issueEventBuilder);
+  }
+
+  /** Returns flow/job metadata tags from job properties for timing event metadata. */
+  private Map<String, String> getFlowMetadata() {
+    Map<String, String> metadata = Maps.newHashMap();
+    metadata.put(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD,
+        this.gobblinProps.getProperty(ConfigurationKeys.FLOW_GROUP_KEY, ""));
+    metadata.put(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD,
+        this.gobblinProps.getProperty(ConfigurationKeys.FLOW_NAME_KEY, ""));
+    metadata.put(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD,
+        this.gobblinProps.getProperty(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, ""));
+    metadata.put(TimingEvent.FlowEventConstants.JOB_GROUP_FIELD,
+        this.gobblinProps.getProperty(ConfigurationKeys.JOB_GROUP_KEY, ""));
+    metadata.put(TimingEvent.FlowEventConstants.JOB_NAME_FIELD,
+        this.gobblinProps.getProperty(ConfigurationKeys.JOB_NAME_KEY, ""));
+    return metadata;
+  }
 }

gobblin-modules/gobblin-azkaban/src/test/java/org/apache/gobblin/azkaban/AzkabanGobblinYarnAppLauncherTest.java

@@ -0,0 +1,133 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.gobblin.azkaban;
+
+import java.io.IOException;
+import java.util.Properties;
+
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+import org.apache.gobblin.configuration.ConfigurationKeys;
+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.runtime.app.ApplicationLauncher;
+import org.apache.gobblin.runtime.troubleshooter.IssueEventBuilder;
+import org.apache.gobblin.yarn.GobblinYarnAppLauncher;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+
+@Test
+public class AzkabanGobblinYarnAppLauncherTest {
+
+  private static Properties buildProps() {
+    Properties props = new Properties();
+    props.setProperty(ConfigurationKeys.FLOW_GROUP_KEY, "testGroup");
+    props.setProperty(ConfigurationKeys.FLOW_NAME_KEY, "testFlow");
+    props.setProperty(ConfigurationKeys.FLOW_EXECUTION_ID_KEY, "12345");
+    props.setProperty(ConfigurationKeys.JOB_NAME_KEY, "testJob");
+    props.setProperty(ConfigurationKeys.JOB_GROUP_KEY, "testGroup");
+    return props;
+  }
+
+  private AzkabanGobblinYarnAppLauncher buildLauncher(GobblinYarnAppLauncher yarnLauncher,
+      EventSubmitter eventSubmitter) throws IOException {
+    ApplicationLauncher appLauncher = mock(ApplicationLauncher.class);
+    return new AzkabanGobblinYarnAppLauncher("testJob", buildProps(), yarnLauncher, appLauncher, eventSubmitter);
+  }
+
+  @Test
+  public void runCallsSubmitJobFailedEventOnLaunchFailure() throws Exception {
+    GobblinYarnAppLauncher yarnLauncher = mock(GobblinYarnAppLauncher.class);
+    EventSubmitter eventSubmitter = mock(EventSubmitter.class);
+    TimingEvent mockTimer = mock(TimingEvent.class);
+    Mockito.when(eventSubmitter.getTimingEvent(anyString())).thenReturn(mockTimer);
+
+    IOException cause = new IOException("token refresh failed");
+    doThrow(cause).when(yarnLauncher).launch();
+
+    AzkabanGobblinYarnAppLauncher launcher = buildLauncher(yarnLauncher, eventSubmitter);
+
+    try {
+      launcher.run();
+      Assert.fail("Expected IOException to propagate");
+    } catch (IOException e) {
+      Assert.assertEquals(e, cause);
+    }
+
+    // Only JOB_FAILED timing event (no JOB_COMPLETE for bootstrap failures)
+    verify(eventSubmitter, times(1)).getTimingEvent(TimingEvent.LauncherTimings.JOB_FAILED);
+    verify(mockTimer, times(1)).stop(any());
+
+    // IssueEventBuilder submitted with bootstrap failure summary
+    ArgumentCaptor<GobblinEventBuilder> captor = ArgumentCaptor.forClass(GobblinEventBuilder.class);
+    verify(eventSubmitter, times(1)).submit(captor.capture());
+    Assert.assertTrue(captor.getValue() instanceof IssueEventBuilder);
+    IssueEventBuilder issueEvent = (IssueEventBuilder) captor.getValue();
+    Assert.assertTrue(issueEvent.getIssue().getSummary().contains("YARN AM bootstrap failed"));
+    Assert.assertTrue(issueEvent.getIssue().getSummary().contains("token refresh failed"));
+  }
+
+  @Test
+  public void runDoesNotCallSubmitJobFailedEventOnSuccess() throws Exception {
+    GobblinYarnAppLauncher yarnLauncher = mock(GobblinYarnAppLauncher.class);
+    EventSubmitter eventSubmitter = mock(EventSubmitter.class);
+
+    // launch() succeeds
+    AzkabanGobblinYarnAppLauncher launcher = buildLauncher(yarnLauncher, eventSubmitter);
+    launcher.run();
+
+    verify(eventSubmitter, times(0)).getTimingEvent(anyString());
+    verify(eventSubmitter, times(0)).submit(any(GobblinEventBuilder.class));
+  }
+
+  @Test
+  public void issueEventContainsFlowMetadata() throws Exception {
+    GobblinYarnAppLauncher yarnLauncher = mock(GobblinYarnAppLauncher.class);
+    EventSubmitter eventSubmitter = mock(EventSubmitter.class);
+    TimingEvent mockTimer = mock(TimingEvent.class);
+    Mockito.when(eventSubmitter.getTimingEvent(anyString())).thenReturn(mockTimer);
+    doThrow(new IOException("hdfs quota exceeded")).when(yarnLauncher).launch();
+
+    AzkabanGobblinYarnAppLauncher launcher = buildLauncher(yarnLauncher, eventSubmitter);
+
+    try {
+      launcher.run();
+    } catch (IOException ignored) {
+    }
+
+    ArgumentCaptor<GobblinEventBuilder> captor = ArgumentCaptor.forClass(GobblinEventBuilder.class);
+    verify(eventSubmitter).submit(captor.capture());
+
+    IssueEventBuilder issue = (IssueEventBuilder) captor.getValue();
+    Assert.assertEquals(issue.getMetadata().get(TimingEvent.FlowEventConstants.FLOW_GROUP_FIELD), "testGroup");
+    Assert.assertEquals(issue.getMetadata().get(TimingEvent.FlowEventConstants.FLOW_NAME_FIELD), "testFlow");
+    Assert.assertEquals(issue.getMetadata().get(TimingEvent.FlowEventConstants.FLOW_EXECUTION_ID_FIELD), "12345");
+    Assert.assertEquals(issue.getMetadata().get(TimingEvent.FlowEventConstants.JOB_NAME_FIELD), "testJob");
+  }
+}