Fix unit test failures from env configs (#10998)

* Fix unit test failures from env configs

* Default null to empty string

* Format code

* Fix one more unit test

* Remove unused import

Author: tuliren

airbyte-commons/src/main/java/io/airbyte/commons/lang/Exceptions.java

@@ -71,4 +71,12 @@ public interface Procedure {
 
   }
 
+  public static <T> T swallowWithDefault(final Callable<T> procedure, final T defaultValue) {
+    try {
+      return procedure.call();
+    } catch (Exception e) {
+      return defaultValue;
+    }
+  }
+
 }

airbyte-config/models/src/main/java/io/airbyte/config/EnvConfigs.java

@@ -7,6 +7,7 @@
 import com.google.common.base.Preconditions;
 import com.google.common.base.Splitter;
 import com.google.common.base.Strings;
+import io.airbyte.commons.lang.Exceptions;
 import io.airbyte.commons.map.MoreMaps;
 import io.airbyte.commons.version.AirbyteVersion;
 import io.airbyte.config.helpers.LogClientSingleton;
@@ -651,9 +652,12 @@ public Map<String, String> getJobDefaultEnvMap() {
     final Map<String, String> jobPrefixedEnvMap = getAllEnvKeys.get().stream()
         .filter(key -> key.startsWith(JOB_DEFAULT_ENV_PREFIX))
         .collect(Collectors.toMap(key -> key.replace(JOB_DEFAULT_ENV_PREFIX, ""), getEnv));
+    // This method assumes that these shared env variables are not critical to the execution
+    // of the jobs, and only serve as metadata. So any exception is swallowed and default to
+    // an empty string. Change this logic if this assumption no longer holds.
     final Map<String, String> jobSharedEnvMap = JOB_SHARED_ENVS.entrySet().stream().collect(Collectors.toMap(
         Entry::getKey,
-        entry -> Objects.requireNonNullElse(entry.getValue().apply(this), "")));
+        entry -> Exceptions.swallowWithDefault(() -> Objects.requireNonNullElse(entry.getValue().apply(this), ""), "")));
     return MoreMaps.merge(jobPrefixedEnvMap, jobSharedEnvMap);
   }
 

airbyte-workers/src/test/java/io/airbyte/workers/process/AirbyteIntegrationLauncherTest.java

@@ -7,6 +7,7 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import io.airbyte.config.EnvConfigs;
+import io.airbyte.config.WorkerEnvConstants;
 import io.airbyte.workers.WorkerConfigs;
 import io.airbyte.workers.WorkerException;
 import java.nio.file.Path;
@@ -31,6 +32,10 @@ class AirbyteIntegrationLauncherTest {
       "config", "{}",
       "catalog", "{}",
       "state", "{}");
+  private static final Map<String, String> JOB_METADATA = Map.of(
+      WorkerEnvConstants.WORKER_CONNECTOR_IMAGE, FAKE_IMAGE,
+      WorkerEnvConstants.WORKER_JOB_ID, JOB_ID,
+      WorkerEnvConstants.WORKER_JOB_ATTEMPT, String.valueOf(JOB_ATTEMPT));
 
   private WorkerConfigs workerConfigs;
   private ProcessFactory processFactory;
@@ -48,7 +53,7 @@ void spec() throws WorkerException {
     launcher.spec(JOB_ROOT);
 
     Mockito.verify(processFactory).create(JOB_ID, JOB_ATTEMPT, JOB_ROOT, FAKE_IMAGE, false, Collections.emptyMap(), null,
-        workerConfigs.getResourceRequirements(), Map.of(KubeProcessFactory.JOB_TYPE, KubeProcessFactory.SPEC_JOB), Map.of(), Map.of(),
+        workerConfigs.getResourceRequirements(), Map.of(KubeProcessFactory.JOB_TYPE, KubeProcessFactory.SPEC_JOB), JOB_METADATA, Map.of(),
         "spec");
   }
 
@@ -59,7 +64,7 @@ void check() throws WorkerException {
     Mockito.verify(processFactory).create(JOB_ID, JOB_ATTEMPT, JOB_ROOT, FAKE_IMAGE, false, CONFIG_FILES, null,
         workerConfigs.getResourceRequirements(),
         Map.of(KubeProcessFactory.JOB_TYPE, KubeProcessFactory.CHECK_JOB),
-        Map.of(),
+        JOB_METADATA,
         Map.of(),
         "check",
         "--config", "config");
@@ -72,7 +77,7 @@ void discover() throws WorkerException {
     Mockito.verify(processFactory).create(JOB_ID, JOB_ATTEMPT, JOB_ROOT, FAKE_IMAGE, false, CONFIG_FILES, null,
         workerConfigs.getResourceRequirements(),
         Map.of(KubeProcessFactory.JOB_TYPE, KubeProcessFactory.DISCOVER_JOB),
-        Map.of(),
+        JOB_METADATA,
         Map.of(),
         "discover",
         "--config", "config");
@@ -85,7 +90,7 @@ void read() throws WorkerException {
     Mockito.verify(processFactory).create(JOB_ID, JOB_ATTEMPT, JOB_ROOT, FAKE_IMAGE, false, CONFIG_CATALOG_STATE_FILES, null,
         workerConfigs.getResourceRequirements(),
         Map.of(KubeProcessFactory.JOB_TYPE, KubeProcessFactory.SYNC_JOB, KubeProcessFactory.SYNC_STEP, KubeProcessFactory.READ_STEP),
-        Map.of(),
+        JOB_METADATA,
         Map.of(),
         Lists.newArrayList(
             "read",
@@ -101,7 +106,7 @@ void write() throws WorkerException {
     Mockito.verify(processFactory).create(JOB_ID, JOB_ATTEMPT, JOB_ROOT, FAKE_IMAGE, true, CONFIG_CATALOG_FILES, null,
         workerConfigs.getResourceRequirements(),
         Map.of(KubeProcessFactory.JOB_TYPE, KubeProcessFactory.SYNC_JOB, KubeProcessFactory.SYNC_STEP, KubeProcessFactory.WRITE_STEP),
-        Map.of(),
+        JOB_METADATA,
         Map.of(),
         "write",
         "--config", "config",