[Connector API] Implement update scheduling action (#102799)

Add endpoint to update the connector scheduling.

rest-api-spec/src/main/resources/rest-api-spec/api/connector.update_scheduling.json

@@ -0,0 +1,39 @@
+{
+  "connector.update_scheduling": {
+    "documentation": {
+      "url": "https://www.elastic.co/guide/en/enterprise-search/current/connectors.html",
+      "description": "Updates the scheduling field in the connector document."
+    },
+    "stability": "experimental",
+    "visibility": "feature_flag",
+    "feature_flag": "es.connector_api_feature_flag_enabled",
+    "headers": {
+      "accept": [
+        "application/json"
+      ],
+      "content_type": [
+        "application/json"
+      ]
+    },
+    "url": {
+      "paths": [
+        {
+          "path": "/_connector/{connector_id}/_scheduling",
+          "methods": [
+            "PUT"
+          ],
+          "parts": {
+            "connector_id": {
+              "type": "string",
+              "description": "The unique identifier of the connector to be updated."
+            }
+          }
+        }
+      ]
+    },
+    "body": {
+      "description": "An object containing the connector's scheduling configuration.",
+      "required": true
+    }
+  }
+}

x-pack/plugin/ent-search/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/entsearch/331_connector_update_scheduling.yml

@@ -0,0 +1,94 @@
+setup:
+  - skip:
+      version: " - 8.11.99"
+      reason: Introduced in 8.12.0
+
+  - do:
+      connector.put:
+        connector_id: test-connector
+        body:
+          index_name: search-1-test
+          name: my-connector
+          language: pl
+          is_native: false
+          service_type: super-connector
+
+---
+"Update Connector Scheduling":
+  - do:
+      connector.update_scheduling:
+        connector_id: test-connector
+        body:
+          scheduling:
+            access_control:
+              enabled: true
+              interval: 1 0 0 * * ?
+            full:
+              enabled: false
+              interval: 2 0 0 * * ?
+            incremental:
+              enabled: false
+              interval: 3 0 0 * * ?
+
+
+  - match: { result: updated }
+
+  - do:
+      connector.get:
+        connector_id: test-connector
+
+  - match: { scheduling.access_control.enabled: true }
+  - match: { scheduling.access_control.interval: "1 0 0 * * ?" }
+  - match: { scheduling.full.enabled: false }
+  - match: { scheduling.full.interval: "2 0 0 * * ?" }
+  - match: { scheduling.incremental.enabled: false }
+  - match: { scheduling.incremental.interval: "3 0 0 * * ?" }
+
+---
+"Update Connector Scheduling - 404 status code returned when connector doesn't exist":
+  - do:
+      catch: "missing"
+      connector.update_scheduling:
+        connector_id: test-non-existent-connector
+        body:
+          scheduling:
+            access_control:
+              enabled: true
+              interval: 1 0 0 * * ?
+            full:
+              enabled: false
+              interval: 2 0 0 * * ?
+            incremental:
+              enabled: false
+              interval: 3 0 0 * * ?
+
+---
+"Update Connector Scheduling - 400 status code returned when required fields are missing":
+  - do:
+      catch: "bad_request"
+      connector.update_scheduling:
+        connector_id: test-connector
+        body:
+          scheduling:
+            incremental:
+              enabled: false
+              interval: 3 0 0 * * ?
+
+---
+"Update Connector Scheduling - 400 status code returned with wrong CRON expression":
+  - do:
+      catch: "bad_request"
+      connector.update_scheduling:
+        connector_id: test-connector
+        body:
+          scheduling:
+            access_control:
+              enabled: true
+              interval: 61 0 0 * * ?
+            full:
+              enabled: false
+              interval: 2 0 0 * * ?
+            incremental:
+              enabled: false
+              interval: 3 0 0 * * ?
+

x-pack/plugin/ent-search/src/main/java/org/elasticsearch/xpack/application/EnterpriseSearch.java

@@ -50,10 +50,13 @@ import org.elasticsearch.xpack.application.connector.action.RestDeleteConnectorA
 import org.elasticsearch.xpack.application.connector.action.RestGetConnectorAction;
 import org.elasticsearch.xpack.application.connector.action.RestListConnectorAction;
 import org.elasticsearch.xpack.application.connector.action.RestPutConnectorAction;
+import org.elasticsearch.xpack.application.connector.action.RestUpdateConnectorSchedulingAction;
 import org.elasticsearch.xpack.application.connector.action.TransportDeleteConnectorAction;
 import org.elasticsearch.xpack.application.connector.action.TransportGetConnectorAction;
 import org.elasticsearch.xpack.application.connector.action.TransportListConnectorAction;
 import org.elasticsearch.xpack.application.connector.action.TransportPutConnectorAction;
+import org.elasticsearch.xpack.application.connector.action.TransportUpdateConnectorSchedulingAction;
+import org.elasticsearch.xpack.application.connector.action.UpdateConnectorSchedulingAction;
 import org.elasticsearch.xpack.application.connector.syncjob.action.PostConnectorSyncJobAction;
 import org.elasticsearch.xpack.application.connector.syncjob.action.RestPostConnectorSyncJobAction;
 import org.elasticsearch.xpack.application.connector.syncjob.action.TransportPostConnectorSyncJobAction;
@@ -170,13 +173,17 @@ public class EnterpriseSearch extends Plugin implements ActionPlugin, SystemInde
         if (ConnectorAPIFeature.isEnabled()) {
             actionHandlers.addAll(
                 List.of(
+                    // Connector API
                     new ActionHandler<>(DeleteConnectorAction.INSTANCE, TransportDeleteConnectorAction.class),
                     new ActionHandler<>(GetConnectorAction.INSTANCE, TransportGetConnectorAction.class),
                     new ActionHandler<>(ListConnectorAction.INSTANCE, TransportListConnectorAction.class),
-                    new ActionHandler<>(PutConnectorAction.INSTANCE, TransportPutConnectorAction.class)
+                    new ActionHandler<>(PutConnectorAction.INSTANCE, TransportPutConnectorAction.class),
+                    new ActionHandler<>(UpdateConnectorSchedulingAction.INSTANCE, TransportUpdateConnectorSchedulingAction.class),
+
+                    // SyncJob API
+                    new ActionHandler<>(PostConnectorSyncJobAction.INSTANCE, TransportPostConnectorSyncJobAction.class)
                 )
             );
-            actionHandlers.add(new ActionHandler<>(PostConnectorSyncJobAction.INSTANCE, TransportPostConnectorSyncJobAction.class));
         }
 
         return Collections.unmodifiableList(actionHandlers);
@@ -225,13 +232,17 @@ public class EnterpriseSearch extends Plugin implements ActionPlugin, SystemInde
         if (ConnectorAPIFeature.isEnabled()) {
             restHandlers.addAll(
                 List.of(
+                    // Connector API
                     new RestDeleteConnectorAction(),
                     new RestGetConnectorAction(),
                     new RestListConnectorAction(),
-                    new RestPutConnectorAction()
+                    new RestPutConnectorAction(),
+                    new RestUpdateConnectorSchedulingAction(),
+
+                    // SyncJob API
+                    new RestPostConnectorSyncJobAction()
                 )
             );
-            restHandlers.add(new RestPostConnectorSyncJobAction());
         }
 
         return Collections.unmodifiableList(restHandlers);

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

@@ -201,7 +201,7 @@ public class Connector implements NamedWriteable, ToXContentObject {
     public static final ParseField LANGUAGE_FIELD = new ParseField("language");
     static final ParseField NAME_FIELD = new ParseField("name");
     public static final ParseField PIPELINE_FIELD = new ParseField("pipeline");
-    static final ParseField SCHEDULING_FIELD = new ParseField("scheduling");
+    public static final ParseField SCHEDULING_FIELD = new ParseField("scheduling");
     public static final ParseField SERVICE_TYPE_FIELD = new ParseField("service_type");
     static final ParseField STATUS_FIELD = new ParseField("status");
     static final ParseField SYNC_CURSOR_FIELD = new ParseField("sync_cursor");
@@ -447,6 +447,10 @@ public class Connector implements NamedWriteable, ToXContentObject {
         return connectorId;
     }
 
+    public ConnectorScheduling getScheduling() {
+        return scheduling;
+    }
+
     public List<ConnectorFiltering> getFiltering() {
         return filtering;
     }

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

@@ -20,6 +20,8 @@ import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.search.SearchRequest;
 import org.elasticsearch.action.search.SearchResponse;
 import org.elasticsearch.action.support.WriteRequest;
+import org.elasticsearch.action.update.UpdateRequest;
+import org.elasticsearch.action.update.UpdateResponse;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.client.internal.OriginSettingClient;
 import org.elasticsearch.index.IndexNotFoundException;
@@ -29,6 +31,7 @@ import org.elasticsearch.search.builder.SearchSourceBuilder;
 import org.elasticsearch.search.sort.SortOrder;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentType;
+import org.elasticsearch.xpack.application.connector.action.UpdateConnectorSchedulingAction;
 
 import java.util.Arrays;
 import java.util.Collections;
@@ -166,6 +169,36 @@ public class ConnectorIndexService {
         }
     }
 
+    /**
+     * Updates the {@link ConnectorScheduling} property of a {@link Connector}.
+     *
+     * @param request  The request for updating the connector's scheduling.
+     * @param listener The listener for handling responses, including successful updates or errors.
+     */
+    public void updateConnectorScheduling(UpdateConnectorSchedulingAction.Request request, ActionListener<UpdateResponse> listener) {
+        try {
+            String connectorId = request.getConnectorId();
+            final UpdateRequest updateRequest = new UpdateRequest(CONNECTOR_INDEX_NAME, connectorId).doc(
+                new IndexRequest(CONNECTOR_INDEX_NAME).opType(DocWriteRequest.OpType.INDEX)
+                    .id(connectorId)
+                    .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
+                    .source(request.toXContent(jsonBuilder(), ToXContent.EMPTY_PARAMS))
+            );
+            clientWithOrigin.update(
+                updateRequest,
+                new DelegatingIndexNotFoundActionListener<>(connectorId, listener, (l, updateResponse) -> {
+                    if (updateResponse.getResult() == UpdateResponse.Result.NOT_FOUND) {
+                        l.onFailure(new ResourceNotFoundException(connectorId));
+                        return;
+                    }
+                    l.onResponse(updateResponse);
+                })
+            );
+        } catch (Exception e) {
+            listener.onFailure(e);
+        }
+    }
+
     private static ConnectorIndexService.ConnectorResult mapSearchResponseToConnectorList(SearchResponse response) {
         final List<Connector> connectorResults = Arrays.stream(response.getHits().getHits())
             .map(ConnectorIndexService::hitToConnector)

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

@@ -0,0 +1,45 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.application.connector.action;
+
+import org.elasticsearch.client.internal.node.NodeClient;
+import org.elasticsearch.rest.BaseRestHandler;
+import org.elasticsearch.rest.RestRequest;
+import org.elasticsearch.rest.action.RestToXContentListener;
+import org.elasticsearch.xpack.application.EnterpriseSearch;
+
+import java.util.List;
+
+import static org.elasticsearch.rest.RestRequest.Method.PUT;
+
+public class RestUpdateConnectorSchedulingAction extends BaseRestHandler {
+
+    @Override
+    public String getName() {
+        return "connector_update_scheduling_action";
+    }
+
+    @Override
+    public List<Route> routes() {
+        return List.of(new Route(PUT, "/" + EnterpriseSearch.CONNECTOR_API_ENDPOINT + "/{connector_id}/_scheduling"));
+    }
+
+    @Override
+    protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) {
+        UpdateConnectorSchedulingAction.Request request = UpdateConnectorSchedulingAction.Request.fromXContentBytes(
+            restRequest.param("connector_id"),
+            restRequest.content(),
+            restRequest.getXContentType()
+        );
+        return channel -> client.execute(
+            UpdateConnectorSchedulingAction.INSTANCE,
+            request,
+            new RestToXContentListener<>(channel, UpdateConnectorSchedulingAction.Response::status, r -> null)
+        );
+    }
+}

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

@@ -0,0 +1,55 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.application.connector.action;
+
+import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.support.ActionFilters;
+import org.elasticsearch.action.support.HandledTransportAction;
+import org.elasticsearch.client.internal.Client;
+import org.elasticsearch.cluster.service.ClusterService;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.util.concurrent.EsExecutors;
+import org.elasticsearch.tasks.Task;
+import org.elasticsearch.transport.TransportService;
+import org.elasticsearch.xpack.application.connector.ConnectorIndexService;
+
+public class TransportUpdateConnectorSchedulingAction extends HandledTransportAction<
+    UpdateConnectorSchedulingAction.Request,
+    UpdateConnectorSchedulingAction.Response> {
+
+    protected final ConnectorIndexService connectorIndexService;
+
+    @Inject
+    public TransportUpdateConnectorSchedulingAction(
+        TransportService transportService,
+        ClusterService clusterService,
+        ActionFilters actionFilters,
+        Client client
+    ) {
+        super(
+            UpdateConnectorSchedulingAction.NAME,
+            transportService,
+            actionFilters,
+            UpdateConnectorSchedulingAction.Request::new,
+            EsExecutors.DIRECT_EXECUTOR_SERVICE
+        );
+        this.connectorIndexService = new ConnectorIndexService(client);
+    }
+
+    @Override
+    protected void doExecute(
+        Task task,
+        UpdateConnectorSchedulingAction.Request request,
+        ActionListener<UpdateConnectorSchedulingAction.Response> listener
+    ) {
+        connectorIndexService.updateConnectorScheduling(
+            request,
+            listener.map(r -> new UpdateConnectorSchedulingAction.Response(r.getResult()))
+        );
+    }
+}

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

@@ -0,0 +1,177 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.application.connector.action;
+
+import org.elasticsearch.ElasticsearchParseException;
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.ActionResponse;
+import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.common.bytes.BytesReference;
+import org.elasticsearch.common.io.stream.StreamInput;
+import org.elasticsearch.common.io.stream.StreamOutput;
+import org.elasticsearch.common.xcontent.XContentHelper;
+import org.elasticsearch.rest.RestStatus;
+import org.elasticsearch.xcontent.ConstructingObjectParser;
+import org.elasticsearch.xcontent.ToXContentObject;
+import org.elasticsearch.xcontent.XContentBuilder;
+import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xcontent.XContentParserConfiguration;
+import org.elasticsearch.xcontent.XContentType;
+import org.elasticsearch.xpack.application.connector.Connector;
+import org.elasticsearch.xpack.application.connector.ConnectorScheduling;
+
+import java.io.IOException;
+import java.util.Objects;
+
+import static org.elasticsearch.xcontent.ConstructingObjectParser.constructorArg;
+
+public class UpdateConnectorSchedulingAction extends ActionType<UpdateConnectorSchedulingAction.Response> {
+
+    public static final UpdateConnectorSchedulingAction INSTANCE = new UpdateConnectorSchedulingAction();
+    public static final String NAME = "cluster:admin/xpack/connector/update_scheduling";
+
+    public UpdateConnectorSchedulingAction() {
+        super(NAME, UpdateConnectorSchedulingAction.Response::new);
+    }
+
+    public static class Request extends ActionRequest implements ToXContentObject {
+
+        private final String connectorId;
+        private final ConnectorScheduling scheduling;
+
+        public Request(String connectorId, ConnectorScheduling scheduling) {
+            this.connectorId = connectorId;
+            this.scheduling = scheduling;
+        }
+
+        public Request(StreamInput in) throws IOException {
+            super(in);
+            this.connectorId = in.readString();
+            this.scheduling = in.readOptionalWriteable(ConnectorScheduling::new);
+        }
+
+        public String getConnectorId() {
+            return connectorId;
+        }
+
+        public ConnectorScheduling getScheduling() {
+            return scheduling;
+        }
+
+        @Override
+        public ActionRequestValidationException validate() {
+            return null;
+        }
+
+        private static final ConstructingObjectParser<UpdateConnectorSchedulingAction.Request, String> PARSER =
+            new ConstructingObjectParser<>(
+                "connector_update_scheduling_request",
+                false,
+                ((args, connectorId) -> new UpdateConnectorSchedulingAction.Request(connectorId, (ConnectorScheduling) args[0]))
+            );
+
+        static {
+            PARSER.declareObject(constructorArg(), (p, c) -> ConnectorScheduling.fromXContent(p), Connector.SCHEDULING_FIELD);
+        }
+
+        public static UpdateConnectorSchedulingAction.Request fromXContentBytes(
+            String connectorId,
+            BytesReference source,
+            XContentType xContentType
+        ) {
+            try (XContentParser parser = XContentHelper.createParser(XContentParserConfiguration.EMPTY, source, xContentType)) {
+                return UpdateConnectorSchedulingAction.Request.fromXContent(parser, connectorId);
+            } catch (IOException e) {
+                throw new ElasticsearchParseException("Failed to parse: " + source.utf8ToString(), e);
+            }
+        }
+
+        public static UpdateConnectorSchedulingAction.Request fromXContent(XContentParser parser, String connectorId) throws IOException {
+            return PARSER.parse(parser, connectorId);
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            {
+                builder.field(Connector.SCHEDULING_FIELD.getPreferredName(), scheduling);
+            }
+            builder.endObject();
+            return builder;
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            super.writeTo(out);
+            out.writeString(connectorId);
+            out.writeOptionalWriteable(scheduling);
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Request request = (Request) o;
+            return Objects.equals(connectorId, request.connectorId) && Objects.equals(scheduling, request.scheduling);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(connectorId, scheduling);
+        }
+    }
+
+    public static class Response extends ActionResponse implements ToXContentObject {
+
+        final DocWriteResponse.Result result;
+
+        public Response(StreamInput in) throws IOException {
+            super(in);
+            result = DocWriteResponse.Result.readFrom(in);
+        }
+
+        public Response(DocWriteResponse.Result result) {
+            this.result = result;
+        }
+
+        @Override
+        public void writeTo(StreamOutput out) throws IOException {
+            this.result.writeTo(out);
+        }
+
+        @Override
+        public XContentBuilder toXContent(XContentBuilder builder, Params params) throws IOException {
+            builder.startObject();
+            builder.field("result", this.result.getLowercase());
+            builder.endObject();
+            return builder;
+        }
+
+        public RestStatus status() {
+            return switch (result) {
+                case NOT_FOUND -> RestStatus.NOT_FOUND;
+                default -> RestStatus.OK;
+            };
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Response that = (Response) o;
+            return Objects.equals(result, that.result);
+        }
+
+        @Override
+        public int hashCode() {
+            return Objects.hash(result);
+        }
+    }
+}

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

@@ -11,8 +11,10 @@ import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.DocWriteResponse;
 import org.elasticsearch.action.delete.DeleteResponse;
+import org.elasticsearch.action.update.UpdateResponse;
 import org.elasticsearch.rest.RestStatus;
 import org.elasticsearch.test.ESSingleNodeTestCase;
+import org.elasticsearch.xpack.application.connector.action.UpdateConnectorSchedulingAction;
 import org.junit.Before;
 
 import java.util.ArrayList;
@@ -36,7 +38,6 @@ public class ConnectorIndexServiceTests extends ESSingleNodeTestCase {
     }
 
     public void testPutConnector() throws Exception {
-
         Connector connector = ConnectorTestUtils.getRandomConnector();
         DocWriteResponse resp = awaitPutConnector(connector);
         assertThat(resp.status(), anyOf(equalTo(RestStatus.CREATED), equalTo(RestStatus.OK)));
@@ -60,6 +61,25 @@ public class ConnectorIndexServiceTests extends ESSingleNodeTestCase {
         expectThrows(ResourceNotFoundException.class, () -> awaitDeleteConnector(connectorIdToDelete));
     }
 
+    public void testUpdateConnectorScheduling() throws Exception {
+        Connector connector = ConnectorTestUtils.getRandomConnector();
+        DocWriteResponse resp = awaitPutConnector(connector);
+        assertThat(resp.status(), anyOf(equalTo(RestStatus.CREATED), equalTo(RestStatus.OK)));
+
+        ConnectorScheduling updatedScheduling = ConnectorTestUtils.getRandomConnectorScheduling();
+
+        UpdateConnectorSchedulingAction.Request updateSchedulingRequest = new UpdateConnectorSchedulingAction.Request(
+            connector.getConnectorId(),
+            updatedScheduling
+        );
+
+        DocWriteResponse updateResponse = awaitUpdateConnectorScheduling(updateSchedulingRequest);
+        assertThat(updateResponse.status(), equalTo(RestStatus.OK));
+
+        Connector indexedConnector = awaitGetConnector(connector.getConnectorId());
+        assertThat(updatedScheduling, equalTo(indexedConnector.getScheduling()));
+    }
+
     private DeleteResponse awaitDeleteConnector(String connectorId) throws Exception {
         CountDownLatch latch = new CountDownLatch(1);
         final AtomicReference<DeleteResponse> resp = new AtomicReference<>(null);
@@ -160,4 +180,28 @@ public class ConnectorIndexServiceTests extends ESSingleNodeTestCase {
         return resp.get();
     }
 
+    private UpdateResponse awaitUpdateConnectorScheduling(UpdateConnectorSchedulingAction.Request updatedScheduling) throws Exception {
+        CountDownLatch latch = new CountDownLatch(1);
+        final AtomicReference<UpdateResponse> resp = new AtomicReference<>(null);
+        final AtomicReference<Exception> exc = new AtomicReference<>(null);
+        connectorIndexService.updateConnectorScheduling(updatedScheduling, new ActionListener<>() {
+            @Override
+            public void onResponse(UpdateResponse indexResponse) {
+                resp.set(indexResponse);
+                latch.countDown();
+            }
+
+            @Override
+            public void onFailure(Exception e) {
+                exc.set(e);
+                latch.countDown();
+            }
+        });
+        assertTrue("Timeout waiting for update scheduling request", latch.await(REQUEST_TIMEOUT_SECONDS, TimeUnit.SECONDS));
+        if (exc.get() != null) {
+            throw exc.get();
+        }
+        assertNotNull("Received null response from update scheduling request", resp.get());
+        return resp.get();
+    }
 }

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

@@ -195,7 +195,7 @@ public final class ConnectorTestUtils {
             .setPipeline(randomBoolean() ? getRandomConnectorIngestPipeline() : null)
             .setScheduling(randomBoolean() ? getRandomConnectorScheduling() : null)
             .setStatus(getRandomConnectorStatus())
-            .setSyncCursor(randomBoolean() ? Map.of("foo", "bar") : null)
+            .setSyncCursor(randomBoolean() ? Map.of(randomAlphaOfLengthBetween(5, 10), randomAlphaOfLengthBetween(5, 10)) : null)
             .setSyncNow(randomBoolean())
             .build();
     }

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

@@ -0,0 +1,51 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.application.connector.action;
+
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.xcontent.XContentParser;
+import org.elasticsearch.xpack.application.connector.ConnectorTestUtils;
+import org.elasticsearch.xpack.core.ml.AbstractBWCSerializationTestCase;
+
+import java.io.IOException;
+
+public class UpdateConnectorSchedulingActionRequestBWCSerializingTests extends AbstractBWCSerializationTestCase<
+    UpdateConnectorSchedulingAction.Request> {
+
+    private String connectorId;
+
+    @Override
+    protected Writeable.Reader<UpdateConnectorSchedulingAction.Request> instanceReader() {
+        return UpdateConnectorSchedulingAction.Request::new;
+    }
+
+    @Override
+    protected UpdateConnectorSchedulingAction.Request createTestInstance() {
+        this.connectorId = randomUUID();
+        return new UpdateConnectorSchedulingAction.Request(connectorId, ConnectorTestUtils.getRandomConnectorScheduling());
+    }
+
+    @Override
+    protected UpdateConnectorSchedulingAction.Request mutateInstance(UpdateConnectorSchedulingAction.Request instance) throws IOException {
+        return randomValueOtherThan(instance, this::createTestInstance);
+    }
+
+    @Override
+    protected UpdateConnectorSchedulingAction.Request doParseInstance(XContentParser parser) throws IOException {
+        return UpdateConnectorSchedulingAction.Request.fromXContent(parser, this.connectorId);
+    }
+
+    @Override
+    protected UpdateConnectorSchedulingAction.Request mutateInstanceForVersion(
+        UpdateConnectorSchedulingAction.Request instance,
+        TransportVersion version
+    ) {
+        return instance;
+    }
+}

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

@@ -0,0 +1,43 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+package org.elasticsearch.xpack.application.connector.action;
+
+import org.elasticsearch.TransportVersion;
+import org.elasticsearch.action.DocWriteResponse;
+import org.elasticsearch.common.io.stream.Writeable;
+import org.elasticsearch.xpack.core.ml.AbstractBWCWireSerializationTestCase;
+
+import java.io.IOException;
+
+public class UpdateConnectorSchedulingActionResponseBWCSerializingTests extends AbstractBWCWireSerializationTestCase<
+    UpdateConnectorSchedulingAction.Response> {
+
+    @Override
+    protected Writeable.Reader<UpdateConnectorSchedulingAction.Response> instanceReader() {
+        return UpdateConnectorSchedulingAction.Response::new;
+    }
+
+    @Override
+    protected UpdateConnectorSchedulingAction.Response createTestInstance() {
+        return new UpdateConnectorSchedulingAction.Response(randomFrom(DocWriteResponse.Result.values()));
+    }
+
+    @Override
+    protected UpdateConnectorSchedulingAction.Response mutateInstance(UpdateConnectorSchedulingAction.Response instance)
+        throws IOException {
+        return randomValueOtherThan(instance, this::createTestInstance);
+    }
+
+    @Override
+    protected UpdateConnectorSchedulingAction.Response mutateInstanceForVersion(
+        UpdateConnectorSchedulingAction.Response instance,
+        TransportVersion version
+    ) {
+        return instance;
+    }
+}

x-pack/plugin/security/qa/operator-privileges-tests/src/javaRestTest/java/org/elasticsearch/xpack/security/operator/Constants.java

@@ -128,6 +128,7 @@ public class Constants {
         "cluster:admin/xpack/connector/list",
         "cluster:admin/xpack/connector/put",
         "cluster:admin/xpack/connector/sync_job/post",
+        "cluster:admin/xpack/connector/update_scheduling",
         "cluster:admin/xpack/deprecation/info",
         "cluster:admin/xpack/deprecation/nodes/info",
         "cluster:admin/xpack/enrich/delete",