diff --git a/airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/job_factory/OAuthConfigSupplier.java b/airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/job_factory/OAuthConfigSupplier.java
index 4654a9749b2e4..040e20aed6002 100644
--- a/airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/job_factory/OAuthConfigSupplier.java
+++ b/airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/job_factory/OAuthConfigSupplier.java
@@ -34,6 +34,7 @@ public class OAuthConfigSupplier {
   private static final Logger LOGGER = LoggerFactory.getLogger(OAuthConfigSupplier.class);
 
   public static final String PATH_IN_CONNECTOR_CONFIG = "path_in_connector_config";
+  private static final String PROPERTIES = "properties";
   final private ConfigRepository configRepository;
   private final TrackingClient trackingClient;
 
@@ -99,9 +100,16 @@ private static boolean injectOAuthParameters(final String connectorName,
     }
     // TODO: if we write a migration to flatten persisted configs in db, we don't need to flatten
     // here see https://github.com/airbytehq/airbyte/issues/7624
-    final JsonNode flatOAuthParameters = MoreOAuthParameters.flattenOAuthConfig(oAuthParameters);
-    final JsonNode outputSpec = spec.getAdvancedAuth().getOauthConfigSpecification().getCompleteOauthServerOutputSpecification();
     boolean result = false;
+    final JsonNode flatOAuthParameters = MoreOAuthParameters.flattenOAuthConfig(oAuthParameters);
+    final JsonNode outputSpecTop = spec.getAdvancedAuth().getOauthConfigSpecification().getCompleteOauthServerOutputSpecification();
+    final JsonNode outputSpec;
+    if (outputSpecTop.has(PROPERTIES)) {
+      outputSpec = outputSpecTop.get(PROPERTIES);
+    } else {
+      LOGGER.error(String.format("In %s's advanced_auth spec, completeOAuthServerOutputSpecification does not declare properties.", connectorName));
+      return false;
+    }
     for (final String key : Jsons.keys(outputSpec)) {
       final JsonNode node = outputSpec.get(key);
       if (node.getNodeType() == OBJECT) {
diff --git a/airbyte-scheduler/persistence/src/test/java/io/airbyte/scheduler/persistence/job_factory/OAuthConfigSupplierTest.java b/airbyte-scheduler/persistence/src/test/java/io/airbyte/scheduler/persistence/job_factory/OAuthConfigSupplierTest.java
index 6f060d0c032b6..3c375a217cfc3 100644
--- a/airbyte-scheduler/persistence/src/test/java/io/airbyte/scheduler/persistence/job_factory/OAuthConfigSupplierTest.java
+++ b/airbyte-scheduler/persistence/src/test/java/io/airbyte/scheduler/persistence/job_factory/OAuthConfigSupplierTest.java
@@ -37,6 +37,7 @@ public class OAuthConfigSupplierTest {
 
   public static final String API_CLIENT = "api_client";
   public static final String CREDENTIALS = "credentials";
+  public static final String PROPERTIES = "properties";
 
   private ConfigRepository configRepository;
   private TrackingClient trackingClient;
@@ -211,10 +212,10 @@ private static AdvancedAuth createAdvancedAuth() {
     return new AdvancedAuth()
         .withAuthFlowType(AuthFlowType.OAUTH_2_0)
         .withOauthConfigSpecification(new OAuthConfigSpecification()
-            .withCompleteOauthServerOutputSpecification(Jsons.jsonNode(Map.of(
-                API_CLIENT, Map.of(
+            .withCompleteOauthServerOutputSpecification(Jsons.jsonNode(Map.of(PROPERTIES,
+                Map.of(API_CLIENT, Map.of(
                     "type", "string",
-                    OAuthConfigSupplier.PATH_IN_CONNECTOR_CONFIG, List.of(CREDENTIALS, API_CLIENT))))));
+                    OAuthConfigSupplier.PATH_IN_CONNECTOR_CONFIG, List.of(CREDENTIALS, API_CLIENT)))))));
   }
 
   private void setupStandardDefinitionMock(final AdvancedAuth advancedAuth) throws JsonValidationException, ConfigNotFoundException, IOException {