[Connector API] Assert required service_type of connector before starting a sync (#108769)

x-pack/plugin/ent-search/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/entsearch/connector/sync_job/10_connector_sync_job_post.yml

@@ -20,6 +20,15 @@ setup:
           is_native: false
           service_type: super-connector
 
+  - do:
+      connector.put:
+        connector_id: test-connector-no-service-type
+        body:
+          index_name: search-test-2
+          name: my-connector
+          language: de
+          is_native: false
+
 ---
 'Create connector sync job':
   - do:
@@ -307,6 +316,16 @@ setup:
           trigger_method: full
       catch: bad_request
 
+---
+'Create connector sync job with no service type':
+  - do:
+      connector.sync_job_post:
+        body:
+          id: test-connector-no-service-type
+          job_type: full
+          trigger_method: full
+      catch: bad_request
+
 ---
 "Create connector sync job fails for unprivileged user":
   - skip:

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/connector/syncjob/ConnectorSyncJobIndexService.java

@@ -98,7 +98,6 @@ public class ConnectorSyncJobIndexService {
         String connectorId = request.getId();
         try {
             getSyncJobConnectorInfo(connectorId, listener.delegateFailure((l, connector) -> {
-
                 if (Strings.isNullOrEmpty(connector.getIndexName())) {
                     l.onFailure(
                         new ElasticsearchStatusException(
@@ -112,6 +111,19 @@ public class ConnectorSyncJobIndexService {
                     return;
                 }
 
+                if (Strings.isNullOrEmpty(connector.getServiceType())) {
+                    l.onFailure(
+                        new ElasticsearchStatusException(
+                            "Cannot start a sync for connector ["
+                                + connectorId
+                                + "] with [service_type] not defined. Set the service type of your connector "
+                                + "before starting the sync.",
+                            RestStatus.BAD_REQUEST
+                        )
+                    );
+                    return;
+                }
+
                 Instant now = Instant.now();
                 ConnectorSyncJobType jobType = Objects.requireNonNullElse(request.getJobType(), ConnectorSyncJob.DEFAULT_JOB_TYPE);
                 ConnectorSyncJobTriggerMethod triggerMethod = Objects.requireNonNullElse(

x-pack/plugin/ent-search/src/test/java/org/elasticsearch/xpack/application/connector/ConnectorTestUtils.java

@@ -330,6 +330,7 @@ public final class ConnectorTestUtils {
             .setSyncInfo(getRandomConnectorSyncInfo())
             .setName(randomFrom(new String[] { null, randomAlphaOfLength(10) }))
             .setPipeline(randomBoolean() ? getRandomConnectorIngestPipeline() : null)
+            .setServiceType(randomAlphaOfLengthBetween(5, 10))
             .setScheduling(getRandomConnectorScheduling())
             .setStatus(getRandomConnectorInitialStatus())
             .setSyncCursor(randomBoolean() ? Map.of(randomAlphaOfLengthBetween(5, 10), randomAlphaOfLengthBetween(5, 10)) : null)
@@ -344,6 +345,10 @@ public final class ConnectorTestUtils {
         return getRandomConnectorBuilder().setIndexName(null).build();
     }
 
+    public static Connector getRandomConnectorWithServiceTypeNotDefined() {
+        return getRandomConnectorBuilder().setServiceType(null).build();
+    }
+
     private static BytesReference convertConnectorToBytesReference(Connector connector) {
         try {
             return XContentHelper.toXContent((builder, params) -> {

x-pack/plugin/ent-search/src/test/java/org/elasticsearch/xpack/application/connector/syncjob/ConnectorSyncJobIndexServiceTests.java

@@ -77,6 +77,7 @@ public class ConnectorSyncJobIndexServiceTests extends ESSingleNodeTestCase {
     private String connectorOneId;
     private String connectorTwoId;
     private String connectorThreeId;
+    private String connectorFourId;
 
     @Override
     protected Collection<Class<? extends Plugin>> getPlugins() {
@@ -94,6 +95,7 @@ public class ConnectorSyncJobIndexServiceTests extends ESSingleNodeTestCase {
         connectorOneId = createConnector(ConnectorTestUtils.getRandomConnector());
         connectorTwoId = createConnector(ConnectorTestUtils.getRandomConnector());
         connectorThreeId = createConnector(ConnectorTestUtils.getRandomConnectorWithDetachedIndex());
+        connectorFourId = createConnector(ConnectorTestUtils.getRandomConnectorWithServiceTypeNotDefined());
 
         this.connectorSyncJobIndexService = new ConnectorSyncJobIndexService(client());
     }
@@ -176,6 +178,15 @@ public class ConnectorSyncJobIndexServiceTests extends ESSingleNodeTestCase {
         expectThrows(ElasticsearchStatusException.class, () -> awaitPutConnectorSyncJob(syncJobRequest));
     }
 
+    public void testDeleteConnectorSyncJob_WithServiceTypeNotDefined_ExpectException() {
+        PostConnectorSyncJobAction.Request syncJobRequest = new PostConnectorSyncJobAction.Request(
+            connectorFourId,
+            ConnectorSyncJobType.FULL,
+            ConnectorSyncJobTriggerMethod.ON_DEMAND
+        );
+        expectThrows(ElasticsearchStatusException.class, () -> awaitPutConnectorSyncJob(syncJobRequest));
+    }
+
     public void testDeleteConnectorSyncJob_WithNonExistentConnectorId_ExpectException() {
         PostConnectorSyncJobAction.Request syncJobRequest = new PostConnectorSyncJobAction.Request(
             "non-existent-connector-id",