[Connector API] Support soft-deletes of connectors (#118669)

* [Connector API] Add interface for soft-deletes

* Define connector deleted system index

* Got soft-delete logic working

* Add unit tests

* Add yaml e2e test and attempt to update permissions

* Fix permissions

* Update docs

* Fix docs

* Update docs/changelog/118282.yaml

* Change logic

* Fix tests

* Remove unnecessary privilege from yaml rest test

* Update changelog

* Update docs/changelog/118669.yaml

* Adapt yaml tests

* Undo changes to muted-tests.yml

* Fix compilation issue after other PR got merged

* Exclude soft-deleted connector from checks about index_name already in use

* Update docs/reference/connector/apis/get-connector-api.asciidoc

Co-authored-by: Tim Grein <tim@4greins.de>

* Update rest-api-spec/src/main/resources/rest-api-spec/api/connector.list.json

Co-authored-by: Tim Grein <tim@4greins.de>

* Adapt comments, add connector wire serializing test

* Introduce new transport versions for passing the delete flag

* Get rid of wire serialisation, use include_deleted instead of deleted flag

* Remove unused import

* Final tweaks

* Adapt variable name in rest layer

---------

Co-authored-by: Elastic Machine <elasticmachine@users.noreply.github.com>
Co-authored-by: Tim Grein <tim@4greins.de>

docs/changelog/118669.yaml

@@ -0,0 +1,5 @@
+pr: 118669
+summary: "[Connector API] Support soft-deletes of connectors"
+area: Extract&Transform
+type: feature
+issues: []

docs/reference/connector/apis/delete-connector-api.asciidoc

@@ -6,14 +6,14 @@
 
 beta::[]
 
+
 .New API reference
 [sidebar]
 --
 For the most up-to-date API details, refer to {api-es}/group/endpoint-connector[Connector APIs].
 --
 
-Removes a connector and associated sync jobs.
+Soft-deletes a connector and removes associated sync jobs.
-This is a destructive action that is not recoverable.
 
 Note: this action doesn't delete any API key, ingest pipeline or data index associated with the connector. These need to be removed manually.
 

docs/reference/connector/apis/get-connector-api.asciidoc

@@ -33,6 +33,9 @@ To get started with Connector APIs, check out <<es-connectors-tutorial-api, our
 `<connector_id>`::
 (Required, string)
 
+`include_deleted`::
+(Optional, boolean) A flag indicating whether to also return connectors that have been soft-deleted. Defaults to `false`.
+
 [[get-connector-api-response-codes]]
 ==== {api-response-codes-title}
 

docs/reference/connector/apis/list-connectors-api.asciidoc

@@ -47,6 +47,9 @@ To get started with Connector APIs, check out <<es-connectors-tutorial-api, our
 `service_type`::
 (Optional, string) A comma-separated list of connector service types, used to filter search results.
 
+`include_deleted`::
+(Optional, boolean) A flag indicating whether to also return connectors that have been soft-deleted. Defaults to `false`.
+
 [[list-connector-api-example]]
 ==== {api-examples-title}
 

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

@@ -26,6 +26,13 @@
           }
         }
       ]
+    },
+    "params": {
+      "include_deleted": {
+        "type": "boolean",
+        "default": false,
+        "description": "A flag indicating whether to return connectors that have been soft-deleted."
+      }
     }
   }
 }

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

@@ -47,6 +47,11 @@
       "query": {
         "type": "string",
         "description": "A search string for querying connectors, filtering results by matching against connector names, descriptions, and index names"
+      },
+      "include_deleted": {
+        "type": "boolean",
+        "default": false,
+        "description": "A flag indicating whether to return connectors that have been soft-deleted."
       }
     }
   }

x-pack/plugin/core/template-resources/src/main/resources/entsearch/connector/elastic-connectors-mappings.json

@@ -302,6 +302,9 @@
         },
         "sync_now": {
           "type": "boolean"
+        },
+        "deleted": {
+          "type": "boolean"
         }
       }
     }

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

@@ -218,6 +218,30 @@ setup:
         body:
           index_name: search-test
 
+---
+'Create Connector - Index name used by deleted connector':
+  - do:
+      connector.put:
+        connector_id: test-connector-1
+        body:
+          index_name: search-test
+
+  - match: { result: 'created' }
+
+  - do:
+      connector.delete:
+        connector_id: test-connector-1
+
+  - match: { acknowledged: true }
+
+  - do:
+      connector.put:
+        connector_id: test-connector-2
+        body:
+          index_name: search-test
+
+  - match: { result: 'created' }
+
 ---
 'Create Connector - Without index attached':
   - do:

x-pack/plugin/ent-search/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/entsearch/connector/20_connector_list.yml

@@ -71,7 +71,7 @@ setup:
   - match: { results.1.language: "nl" }
 
 ---
-"List Connector- with size":
+"List Connector - with size":
   - do:
       connector.list:
         size: 2
@@ -106,7 +106,6 @@ setup:
 
   - match: { count: 0 }
 
-
 ---
 "List Connector - filter by index names":
   - do:
@@ -279,4 +278,145 @@ setup:
       connector.list: { }
 
 
+---
+"List Connectors - Soft deleted connectors / no deleted":
+  - requires:
+      cluster_features: ["connector_soft_deletes"]
+      reason: Soft deletes were introduced in 9.0 release
+
+  - do:
+      connector.list:
+        include_deleted: true
+
+  - match: { count: 3 }
+
+
+---
+"List Connectors - Single soft deleted connector":
+  - requires:
+      cluster_features: ["connector_soft_deletes"]
+      reason: Soft deletes were introduced in 9.0 release
+
+  - do:
+      connector.delete:
+        connector_id: connector-a
+
+  - do:
+      connector.list: {}
+
+  - match: { count: 2 }
+
+  - do:
+      connector.list:
+        include_deleted: true
+
+  - match: { count: 3 }
+
+---
+"List Connectors - Soft deleted connectors":
+  - requires:
+      cluster_features: ["connector_soft_deletes"]
+      reason: Soft deletes were introduced in 9.0 release
+
+  - do:
+      connector.delete:
+        connector_id: connector-a
+  - do:
+      connector.delete:
+        connector_id: connector-b
+  - do:
+      connector.delete:
+        connector_id: connector-c
+
+  - do:
+      connector.list:
+        include_deleted: true
+
+  - match: { count: 3 }
+
+  # Alphabetical order by index_name for results
+  - match: { results.0.id: "connector-b" }
+  - match: { results.0.index_name: "content-search-2-test" }
+  - match: { results.0.language: "en" }
+  - match: { results.0.deleted: true }
+
+  - match: { results.1.id: "connector-a" }
+  - match: { results.1.index_name: "search-1-test" }
+  - match: { results.1.language: "pl" }
+  - match: { results.1.deleted: true }
+
+  - match: { results.2.id: "connector-c" }
+  - match: { results.2.index_name: "search-3-test" }
+  - match: { results.2.language: "nl" }
+  - match: { results.2.deleted: true }
+
+
+---
+"List Connectors - Soft deleted with from":
+  - requires:
+      cluster_features: ["connector_soft_deletes"]
+      reason: Soft deletes were introduced in 9.0 release
+
+  - do:
+      connector.delete:
+        connector_id: connector-a
+  - do:
+      connector.delete:
+        connector_id: connector-b
+  - do:
+      connector.delete:
+        connector_id: connector-c
+
+  - do:
+      connector.list:
+        from: 1
+        include_deleted: true
+
+  - match: { count: 3 }
+
+  # Alphabetical order by index_name for results
+  - match: { results.0.id: "connector-a" }
+  - match: { results.0.index_name: "search-1-test" }
+  - match: { results.0.language: "pl" }
+  - match: { results.0.deleted: true }
+
+  - match: { results.1.id: "connector-c" }
+  - match: { results.1.index_name: "search-3-test" }
+  - match: { results.1.language: "nl" }
+  - match: { results.0.deleted: true }
+
+---
+"List Connector - Soft deleted with size":
+  - requires:
+      cluster_features: ["connector_soft_deletes"]
+      reason: Soft deletes were introduced in 9.0 release
+
+  - do:
+      connector.delete:
+        connector_id: connector-a
+  - do:
+      connector.delete:
+        connector_id: connector-b
+  - do:
+      connector.delete:
+        connector_id: connector-c
+
+  - do:
+      connector.list:
+        size: 2
+        include_deleted: true
+
+  - match: { count: 3 }
+
+  # Alphabetical order by index_name for results
+  - match: { results.0.id: "connector-b" }
+  - match: { results.0.index_name: "content-search-2-test" }
+  - match: { results.0.language: "en" }
+  - match: { results.0.deleted: true }
+
+  - match: { results.1.id: "connector-a" }
+  - match: { results.1.index_name: "search-1-test" }
+  - match: { results.1.language: "pl" }
+  - match: { results.1.deleted: true }
+
 

x-pack/plugin/ent-search/qa/rest/src/yamlRestTest/resources/rest-api-spec/test/entsearch/connector/30_connector_delete.yml

@@ -108,6 +108,33 @@ setup:
         connector_id: test-nonexistent-connector
 
 
+---
+"Delete Connector - Supports soft deletes":
+  - requires:
+      cluster_features: ["connector_soft_deletes"]
+      reason: Soft deletes were introduced in 9.0 release
+
+  - do:
+      connector.delete:
+        connector_id: test-connector-to-delete
+
+  - match: { acknowledged: true }
+
+  - do:
+      catch: "missing"
+      connector.get:
+        connector_id: test-connector-to-delete
+
+  - do:
+      connector.get:
+        connector_id: test-connector-to-delete
+        include_deleted: true
+
+  - match: { id: test-connector-to-delete }
+  - match: { index_name: search-1-test }
+  - match: { service_type: super-connector }
+  - match: { name: my-connector }
+
 ---
 "Delete connector fails for unprivileged user":
   - skip:

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

@@ -9,9 +9,6 @@ package org.elasticsearch.xpack.application.connector;
 
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.io.stream.NamedWriteable;
-import org.elasticsearch.common.io.stream.StreamInput;
-import org.elasticsearch.common.io.stream.StreamOutput;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
@@ -63,9 +60,10 @@ import static org.elasticsearch.xpack.application.connector.ConnectorTemplateReg
  *     <li>A {@link ConnectorStatus} indicating the current status of the connector.</li>
  *     <li>A sync cursor, used for incremental syncs.</li>
  *     <li>A boolean flag 'syncNow', which, when set, triggers an immediate synchronization operation.</li>
+ *     <li>A boolean flag 'isDeleted', when set indicates that connector has been soft-deleted. </li>
  * </ul>
  */
-public class Connector implements NamedWriteable, ToXContentObject {
+public class Connector implements ToXContentObject {
 
     public static final String NAME = Connector.class.getName().toUpperCase(Locale.ROOT);
 
@@ -106,6 +104,7 @@ public class Connector implements NamedWriteable, ToXContentObject {
     @Nullable
     private final Object syncCursor;
     private final boolean syncNow;
+    private final boolean isDeleted;
 
     /**
      * Constructor for Connector.
@@ -132,6 +131,7 @@ public class Connector implements NamedWriteable, ToXContentObject {
      * @param status             Current status of the connector.
      * @param syncCursor         Position or state indicating the current point of synchronization.
      * @param syncNow            Flag indicating whether an immediate synchronization is requested.
+     * @param isDeleted          Flag indicating whether connector has been soft-deleted.
      */
     private Connector(
         String connectorId,
@@ -155,7 +155,8 @@ public class Connector implements NamedWriteable, ToXContentObject {
         String serviceType,
         ConnectorStatus status,
         Object syncCursor,
-        boolean syncNow
+        boolean syncNow,
+        Boolean isDeleted
     ) {
         this.connectorId = connectorId;
         this.apiKeyId = apiKeyId;
@@ -179,31 +180,7 @@ public class Connector implements NamedWriteable, ToXContentObject {
         this.status = status;
         this.syncCursor = syncCursor;
         this.syncNow = syncNow;
-    }
+        this.isDeleted = isDeleted;
-
-    public Connector(StreamInput in) throws IOException {
-        this.connectorId = in.readOptionalString();
-        this.apiKeyId = in.readOptionalString();
-        this.apiKeySecretId = in.readOptionalString();
-        this.configuration = in.readMap(ConnectorConfiguration::new);
-        this.customScheduling = in.readMap(ConnectorCustomSchedule::new);
-        this.description = in.readOptionalString();
-        this.error = in.readOptionalString();
-        this.features = in.readOptionalWriteable(ConnectorFeatures::new);
-        this.filtering = in.readOptionalCollectionAsList(ConnectorFiltering::new);
-        this.syncJobFiltering = in.readOptionalWriteable(FilteringRules::new);
-        this.indexName = in.readOptionalString();
-        this.isNative = in.readBoolean();
-        this.language = in.readOptionalString();
-        this.lastSeen = in.readOptionalInstant();
-        this.syncInfo = in.readOptionalWriteable(ConnectorSyncInfo::new);
-        this.name = in.readOptionalString();
-        this.pipeline = in.readOptionalWriteable(ConnectorIngestPipeline::new);
-        this.scheduling = in.readOptionalWriteable(ConnectorScheduling::new);
-        this.serviceType = in.readOptionalString();
-        this.status = in.readEnum(ConnectorStatus.class);
-        this.syncCursor = in.readGenericValue();
-        this.syncNow = in.readBoolean();
     }
 
     public static final ParseField ID_FIELD = new ParseField("id");
@@ -226,6 +203,7 @@ public class Connector implements NamedWriteable, ToXContentObject {
     public static final ParseField STATUS_FIELD = new ParseField("status");
     public static final ParseField SYNC_CURSOR_FIELD = new ParseField("sync_cursor");
     static final ParseField SYNC_NOW_FIELD = new ParseField("sync_now");
+    public static final ParseField IS_DELETED_FIELD = new ParseField("deleted");
 
     @SuppressWarnings("unchecked")
     private static final ConstructingObjectParser<Connector, String> PARSER = new ConstructingObjectParser<>(
@@ -265,7 +243,8 @@ public class Connector implements NamedWriteable, ToXContentObject {
                 .setServiceType((String) args[i++])
                 .setStatus((ConnectorStatus) args[i++])
                 .setSyncCursor(args[i++])
-                .setSyncNow((Boolean) args[i])
+                .setSyncNow((Boolean) args[i++])
+                .setIsDeleted((Boolean) args[i])
                 .build();
         }
     );
@@ -357,6 +336,7 @@ public class Connector implements NamedWriteable, ToXContentObject {
         );
         PARSER.declareObjectOrNull(optionalConstructorArg(), (p, c) -> p.map(), null, SYNC_CURSOR_FIELD);
         PARSER.declareBoolean(optionalConstructorArg(), SYNC_NOW_FIELD);
+        PARSER.declareBoolean(optionalConstructorArg(), IS_DELETED_FIELD);
     }
 
     public static Connector fromXContentBytes(BytesReference source, String docId, XContentType xContentType) {
@@ -433,6 +413,7 @@ public class Connector implements NamedWriteable, ToXContentObject {
             builder.field(STATUS_FIELD.getPreferredName(), status.toString());
         }
         builder.field(SYNC_NOW_FIELD.getPreferredName(), syncNow);
+        builder.field(IS_DELETED_FIELD.getPreferredName(), isDeleted);
     }
 
     @Override
@@ -445,32 +426,6 @@ public class Connector implements NamedWriteable, ToXContentObject {
         return builder;
     }
 
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        out.writeOptionalString(connectorId);
-        out.writeOptionalString(apiKeyId);
-        out.writeOptionalString(apiKeySecretId);
-        out.writeMap(configuration, StreamOutput::writeWriteable);
-        out.writeMap(customScheduling, StreamOutput::writeWriteable);
-        out.writeOptionalString(description);
-        out.writeOptionalString(error);
-        out.writeOptionalWriteable(features);
-        out.writeOptionalCollection(filtering);
-        out.writeOptionalWriteable(syncJobFiltering);
-        out.writeOptionalString(indexName);
-        out.writeBoolean(isNative);
-        out.writeOptionalString(language);
-        out.writeOptionalInstant(lastSeen);
-        out.writeOptionalWriteable(syncInfo);
-        out.writeOptionalString(name);
-        out.writeOptionalWriteable(pipeline);
-        out.writeOptionalWriteable(scheduling);
-        out.writeOptionalString(serviceType);
-        out.writeEnum(status);
-        out.writeGenericValue(syncCursor);
-        out.writeBoolean(syncNow);
-    }
-
     public String getConnectorId() {
         return connectorId;
     }
@@ -559,6 +514,10 @@ public class Connector implements NamedWriteable, ToXContentObject {
         return syncNow;
     }
 
+    public boolean isDeleted() {
+        return isDeleted;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) return true;
@@ -566,6 +525,7 @@ public class Connector implements NamedWriteable, ToXContentObject {
         Connector connector = (Connector) o;
         return isNative == connector.isNative
             && syncNow == connector.syncNow
+            && isDeleted == connector.isDeleted
             && Objects.equals(connectorId, connector.connectorId)
             && Objects.equals(apiKeyId, connector.apiKeyId)
             && Objects.equals(apiKeySecretId, connector.apiKeySecretId)
@@ -612,15 +572,11 @@ public class Connector implements NamedWriteable, ToXContentObject {
             serviceType,
             status,
             syncCursor,
-            syncNow
+            syncNow,
+            isDeleted
         );
     }
 
-    @Override
-    public String getWriteableName() {
-        return NAME;
-    }
-
     public static class Builder {
 
         private String connectorId;
@@ -645,6 +601,7 @@ public class Connector implements NamedWriteable, ToXContentObject {
         private ConnectorStatus status = ConnectorStatus.CREATED;
         private Object syncCursor;
         private boolean syncNow;
+        private boolean isDeleted;
 
         public Builder setConnectorId(String connectorId) {
             this.connectorId = connectorId;
@@ -756,6 +713,11 @@ public class Connector implements NamedWriteable, ToXContentObject {
             return this;
         }
 
+        public Builder setIsDeleted(Boolean isDeleted) {
+            this.isDeleted = Objects.requireNonNullElse(isDeleted, false);
+            return this;
+        }
+
         public Connector build() {
             return new Connector(
                 connectorId,
@@ -779,7 +741,8 @@ public class Connector implements NamedWriteable, ToXContentObject {
                 serviceType,
                 status,
                 syncCursor,
-                syncNow
+                syncNow,
+                isDeleted
             );
         }
     }

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

@@ -13,9 +13,6 @@ import org.elasticsearch.ResourceNotFoundException;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.DelegatingActionListener;
 import org.elasticsearch.action.DocWriteRequest;
-import org.elasticsearch.action.DocWriteResponse;
-import org.elasticsearch.action.delete.DeleteRequest;
-import org.elasticsearch.action.delete.DeleteResponse;
 import org.elasticsearch.action.get.GetRequest;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.search.SearchRequest;
@@ -194,10 +191,13 @@ public class ConnectorIndexService {
     /**
      * Gets the {@link Connector} from the underlying index.
      *
-     * @param connectorId The id of the connector object.
+     * @param connectorId     The id of the connector object.
-     * @param listener    The action listener to invoke on response/failure.
+     * @param includeDeleted  If false, returns only the non-deleted connector with the matching ID;
+     *                        if true, returns the connector with the matching ID.
+     * @param listener        The action listener to invoke on response/failure.
      */
-    public void getConnector(String connectorId, ActionListener<ConnectorSearchResult> listener) {
+    public void getConnector(String connectorId, boolean includeDeleted, ActionListener<ConnectorSearchResult> listener) {
+
         try {
             final GetRequest getRequest = new GetRequest(CONNECTOR_INDEX_NAME).id(connectorId).realtime(true);
 
@@ -212,6 +212,12 @@ public class ConnectorIndexService {
                         .setResultMap(getResponse.getSourceAsMap())
                         .build();
 
+                    boolean connectorIsSoftDeleted = includeDeleted == false && isConnectorDeleted(connector);
+
+                    if (connectorIsSoftDeleted) {
+                        l.onFailure(new ResourceNotFoundException(connectorNotFoundErrorMsg(connectorId)));
+                        return;
+                    }
                     l.onResponse(connector);
                 } catch (Exception e) {
                     listener.onFailure(e);
@@ -223,33 +229,40 @@ public class ConnectorIndexService {
     }
 
     /**
-     * Deletes the {@link Connector} and the related instances of {@link ConnectorSyncJob} in the underlying index.
+     * Soft deletes the {@link Connector} and optionally removes the related instances of {@link ConnectorSyncJob} in the underlying index.
      *
      * @param connectorId          The id of the {@link Connector}.
      * @param shouldDeleteSyncJobs The flag indicating if {@link ConnectorSyncJob} should also be deleted.
      * @param listener             The action listener to invoke on response/failure.
      */
-    public void deleteConnector(String connectorId, boolean shouldDeleteSyncJobs, ActionListener<DeleteResponse> listener) {
+    public void deleteConnector(String connectorId, boolean shouldDeleteSyncJobs, ActionListener<UpdateResponse> listener) {
-
-        final DeleteRequest deleteRequest = new DeleteRequest(CONNECTOR_INDEX_NAME).id(connectorId)
-            .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
 
         try {
-            client.delete(deleteRequest, new DelegatingIndexNotFoundActionListener<>(connectorId, listener, (l, deleteResponse) -> {
+            // ensure that if connector is soft-deleted, deleting it again results in 404
-                if (deleteResponse.getResult() == DocWriteResponse.Result.NOT_FOUND) {
+            getConnector(connectorId, false, listener.delegateFailure((l, connector) -> {
-                    l.onFailure(new ResourceNotFoundException(connectorNotFoundErrorMsg(connectorId)));
+                final UpdateRequest updateRequest = new UpdateRequest(CONNECTOR_INDEX_NAME, connectorId).setRefreshPolicy(
-                    return;
+                    WriteRequest.RefreshPolicy.IMMEDIATE
-                }
+                )
-                if (shouldDeleteSyncJobs) {
+                    .doc(
-                    new ConnectorSyncJobIndexService(client).deleteAllSyncJobsByConnectorId(connectorId, l.map(r -> deleteResponse));
+                        new IndexRequest(CONNECTOR_INDEX_NAME).opType(DocWriteRequest.OpType.INDEX)
-                } else {
+                            .id(connectorId)
-                    l.onResponse(deleteResponse);
+                            .source(Map.of(Connector.IS_DELETED_FIELD.getPreferredName(), true))
-                }
+                    );
+                client.update(updateRequest, new DelegatingIndexNotFoundActionListener<>(connectorId, l, (ll, updateResponse) -> {
+                    if (updateResponse.getResult() == UpdateResponse.Result.NOT_FOUND) {
+                        ll.onFailure(new ResourceNotFoundException(connectorNotFoundErrorMsg(connectorId)));
+                        return;
+                    }
+                    if (shouldDeleteSyncJobs) {
+                        new ConnectorSyncJobIndexService(client).deleteAllSyncJobsByConnectorId(connectorId, ll.map(r -> updateResponse));
+                    } else {
+                        ll.onResponse(updateResponse);
+                    }
+                }));
             }));
         } catch (Exception e) {
             listener.onFailure(e);
         }
-
     }
 
     /**
@@ -261,6 +274,7 @@ public class ConnectorIndexService {
      * @param connectorNames Filter connectors by connector names, if provided.
      * @param serviceTypes Filter connectors by service types, if provided.
      * @param searchQuery Apply a wildcard search on index name, connector name, and description, if provided.
+     * @param includeDeleted  If false, filters to include only non-deleted connectors; if true, no filter is applied.
      * @param listener Invoked with search results or upon failure.
      */
     public void listConnectors(
@@ -270,12 +284,13 @@ public class ConnectorIndexService {
         List<String> connectorNames,
         List<String> serviceTypes,
         String searchQuery,
+        boolean includeDeleted,
         ActionListener<ConnectorIndexService.ConnectorResult> listener
     ) {
         try {
             final SearchSourceBuilder source = new SearchSourceBuilder().from(from)
                 .size(size)
-                .query(buildListQuery(indexNames, connectorNames, serviceTypes, searchQuery))
+                .query(buildListQuery(indexNames, connectorNames, serviceTypes, searchQuery, includeDeleted))
                 .fetchSource(true)
                 .sort(Connector.INDEX_NAME_FIELD.getPreferredName(), SortOrder.ASC);
             final SearchRequest req = new SearchRequest(CONNECTOR_INDEX_NAME).source(source);
@@ -311,13 +326,15 @@ public class ConnectorIndexService {
      * @param connectorNames List of connector names for filtering, or null/empty to skip.
      * @param serviceTypes List of connector service types for filtering, or null/empty to skip.
      * @param searchQuery Search query for wildcard filtering on index name, connector name, and description, or null/empty to skip.
+     * @param includeDeleted  If false, filters to include only non-deleted connectors; if true, no filter is applied.
      * @return A {@link QueryBuilder} customized based on provided filters.
      */
     private QueryBuilder buildListQuery(
         List<String> indexNames,
         List<String> connectorNames,
         List<String> serviceTypes,
-        String searchQuery
+        String searchQuery,
+        boolean includeDeleted
     ) {
         boolean filterByIndexNames = indexNames != null && indexNames.isEmpty() == false;
         boolean filterByConnectorNames = indexNames != null && connectorNames.isEmpty() == false;
@@ -349,7 +366,12 @@ public class ConnectorIndexService {
                     );
             }
         }
-        return usesFilter ? boolFilterQueryBuilder : new MatchAllQueryBuilder();
+
+        if (includeDeleted == false) {
+            boolFilterQueryBuilder.mustNot(new TermQueryBuilder(Connector.IS_DELETED_FIELD.getPreferredName(), true));
+        }
+
+        return boolFilterQueryBuilder;
     }
 
     /**
@@ -368,7 +390,7 @@ public class ConnectorIndexService {
             Map<String, Object> configurationValues = request.getConfigurationValues();
             String connectorId = request.getConnectorId();
 
-            getConnector(connectorId, listener.delegateFailure((l, connector) -> {
+            getConnector(connectorId, false, listener.delegateFailure((l, connector) -> {
 
                 UpdateRequest updateRequest = new UpdateRequest(CONNECTOR_INDEX_NAME, connectorId).setRefreshPolicy(
                     WriteRequest.RefreshPolicy.IMMEDIATE
@@ -600,7 +622,7 @@ public class ConnectorIndexService {
         ActionListener<UpdateResponse> listener
     ) {
         try {
-            getConnector(connectorId, listener.delegateFailure((l, connector) -> {
+            getConnector(connectorId, false, listener.delegateFailure((l, connector) -> {
                 List<ConnectorFiltering> connectorFilteringList = fromXContentBytesConnectorFiltering(
                     connector.getSourceRef(),
                     XContentType.JSON
@@ -660,7 +682,7 @@ public class ConnectorIndexService {
         FilteringValidationInfo validation,
         ActionListener<UpdateResponse> listener
     ) {
-        getConnector(connectorId, listener.delegateFailure((l, connector) -> {
+        getConnector(connectorId, false, listener.delegateFailure((l, connector) -> {
             try {
                 List<ConnectorFiltering> connectorFilteringList = fromXContentBytesConnectorFiltering(
                     connector.getSourceRef(),
@@ -704,7 +726,7 @@ public class ConnectorIndexService {
      * @param listener     Listener to respond to a successful response or an error.
      */
     public void activateConnectorDraftFiltering(String connectorId, ActionListener<UpdateResponse> listener) {
-        getConnector(connectorId, listener.delegateFailure((l, connector) -> {
+        getConnector(connectorId, false, listener.delegateFailure((l, connector) -> {
             try {
                 List<ConnectorFiltering> connectorFilteringList = fromXContentBytesConnectorFiltering(
                     connector.getSourceRef(),
@@ -819,7 +841,7 @@ public class ConnectorIndexService {
             String connectorId = request.getConnectorId();
             boolean isNative = request.isNative();
 
-            getConnector(connectorId, listener.delegateFailure((l, connector) -> {
+            getConnector(connectorId, false, listener.delegateFailure((l, connector) -> {
 
                 String indexName = getConnectorIndexNameFromSearchResult(connector);
 
@@ -930,7 +952,7 @@ public class ConnectorIndexService {
                     return;
                 }
 
-                getConnector(connectorId, l.delegateFailure((ll, connector) -> {
+                getConnector(connectorId, false, l.delegateFailure((ll, connector) -> {
 
                     Boolean isNativeConnector = getConnectorIsNativeFlagFromSearchResult(connector);
                     Boolean doesNotHaveContentPrefix = indexName != null && isValidManagedConnectorIndexName(indexName) == false;
@@ -1013,7 +1035,7 @@ public class ConnectorIndexService {
     public void updateConnectorServiceType(UpdateConnectorServiceTypeAction.Request request, ActionListener<UpdateResponse> listener) {
         try {
             String connectorId = request.getConnectorId();
-            getConnector(connectorId, listener.delegateFailure((l, connector) -> {
+            getConnector(connectorId, false, listener.delegateFailure((l, connector) -> {
 
                 ConnectorStatus prevStatus = getConnectorStatusFromSearchResult(connector);
                 ConnectorStatus newStatus = prevStatus == ConnectorStatus.CREATED
@@ -1060,7 +1082,7 @@ public class ConnectorIndexService {
         try {
             String connectorId = request.getConnectorId();
             ConnectorStatus newStatus = request.getStatus();
-            getConnector(connectorId, listener.delegateFailure((l, connector) -> {
+            getConnector(connectorId, false, listener.delegateFailure((l, connector) -> {
 
                 ConnectorStatus prevStatus = getConnectorStatusFromSearchResult(connector);
 
@@ -1138,6 +1160,11 @@ public class ConnectorIndexService {
         return (Map<String, Object>) searchResult.getResultMap().get(Connector.CONFIGURATION_FIELD.getPreferredName());
     }
 
+    private boolean isConnectorDeleted(ConnectorSearchResult searchResult) {
+        Boolean isDeletedFlag = (Boolean) searchResult.getResultMap().get(Connector.IS_DELETED_FIELD.getPreferredName());
+        return Boolean.TRUE.equals(isDeletedFlag);
+    }
+
     private static ConnectorIndexService.ConnectorResult mapSearchResponseToConnectorList(SearchResponse response) {
         final List<ConnectorSearchResult> connectorResults = Arrays.stream(response.getHits().getHits())
             .map(ConnectorIndexService::hitToConnector)
@@ -1157,7 +1184,7 @@ public class ConnectorIndexService {
 
     /**
      * This method determines if any documents in the connector index have the same index name as the one specified,
-     * excluding the document with the given _id if it is provided.
+     * excluding the docs marked as deleted (soft-deleted) and document with the given _id if it is provided.
      *
      * @param indexName    The name of the index to check for existence in the connector index.
      * @param connectorId  The ID of the {@link Connector} to exclude from the search. Can be null if no document should be excluded.
@@ -1173,6 +1200,9 @@ public class ConnectorIndexService {
 
             boolFilterQueryBuilder.must().add(new TermQueryBuilder(Connector.INDEX_NAME_FIELD.getPreferredName(), indexName));
 
+            // exclude soft-deleted connectors
+            boolFilterQueryBuilder.mustNot(new TermQueryBuilder(Connector.IS_DELETED_FIELD.getPreferredName(), true));
+
             // If we know the connector _id, exclude this from search query
             if (connectorId != null) {
                 boolFilterQueryBuilder.mustNot(new IdsQueryBuilder().addIds(connectorId));

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

@@ -10,6 +10,7 @@ package org.elasticsearch.xpack.application.connector.action;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.support.TransportAction;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -25,6 +26,7 @@ import java.util.Objects;
 
 import static org.elasticsearch.action.ValidateActions.addValidationError;
 import static org.elasticsearch.xcontent.ConstructingObjectParser.constructorArg;
+import static org.elasticsearch.xcontent.ConstructingObjectParser.optionalConstructorArg;
 
 public class GetConnectorAction {
 
@@ -36,22 +38,25 @@ public class GetConnectorAction {
     public static class Request extends ConnectorActionRequest implements ToXContentObject {
 
         private final String connectorId;
+        private final Boolean includeDeleted;
 
         private static final ParseField CONNECTOR_ID_FIELD = new ParseField("connector_id");
 
-        public Request(String connectorId) {
+        private static final ParseField INCLUDE_DELETED_FIELD = new ParseField("include_deleted");
-            this.connectorId = connectorId;
-        }
 
-        public Request(StreamInput in) throws IOException {
+        public Request(String connectorId, Boolean includeDeleted) {
-            super(in);
+            this.connectorId = connectorId;
-            this.connectorId = in.readString();
+            this.includeDeleted = includeDeleted;
         }
 
         public String getConnectorId() {
             return connectorId;
         }
 
+        public Boolean getIncludeDeleted() {
+            return includeDeleted;
+        }
+
         @Override
         public ActionRequestValidationException validate() {
             ActionRequestValidationException validationException = null;
@@ -65,8 +70,7 @@ public class GetConnectorAction {
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            super.writeTo(out);
+            TransportAction.localOnly();
-            out.writeString(connectorId);
         }
 
         @Override
@@ -74,12 +78,12 @@ public class GetConnectorAction {
             if (this == o) return true;
             if (o == null || getClass() != o.getClass()) return false;
             Request request = (Request) o;
-            return Objects.equals(connectorId, request.connectorId);
+            return Objects.equals(connectorId, request.connectorId) && Objects.equals(includeDeleted, request.includeDeleted);
         }
 
         @Override
         public int hashCode() {
-            return Objects.hash(connectorId);
+            return Objects.hash(connectorId, includeDeleted);
         }
 
         @Override
@@ -87,6 +91,7 @@ public class GetConnectorAction {
             builder.startObject();
             {
                 builder.field(CONNECTOR_ID_FIELD.getPreferredName(), connectorId);
+                builder.field(INCLUDE_DELETED_FIELD.getPreferredName(), includeDeleted);
             }
             builder.endObject();
             return builder;
@@ -95,11 +100,12 @@ public class GetConnectorAction {
         private static final ConstructingObjectParser<Request, Void> PARSER = new ConstructingObjectParser<>(
             "get_connector_request",
             false,
-            (p) -> new Request((String) p[0])
+            (p) -> new Request((String) p[0], (Boolean) p[1])
 
         );
         static {
             PARSER.declareString(constructorArg(), CONNECTOR_ID_FIELD);
+            PARSER.declareBoolean(optionalConstructorArg(), INCLUDE_DELETED_FIELD);
         }
 
         public static Request parse(XContentParser parser) {

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

@@ -10,6 +10,7 @@ package org.elasticsearch.xpack.application.connector.action;
 import org.elasticsearch.action.ActionRequestValidationException;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.support.TransportAction;
 import org.elasticsearch.cluster.metadata.MetadataCreateIndexService;
 import org.elasticsearch.common.io.stream.StreamInput;
 import org.elasticsearch.common.io.stream.StreamOutput;
@@ -46,33 +47,28 @@ public class ListConnectorAction {
         private final List<String> connectorNames;
         private final List<String> connectorServiceTypes;
         private final String connectorSearchQuery;
+        private final Boolean includeDeleted;
 
         private static final ParseField PAGE_PARAMS_FIELD = new ParseField("pageParams");
         private static final ParseField INDEX_NAMES_FIELD = new ParseField("index_names");
         private static final ParseField NAMES_FIELD = new ParseField("names");
         private static final ParseField SEARCH_QUERY_FIELD = new ParseField("query");
-
+        private static final ParseField INCLUDE_DELETED_FIELD = new ParseField("include_deleted");
-        public Request(StreamInput in) throws IOException {
-            super(in);
-            this.pageParams = new PageParams(in);
-            this.indexNames = in.readOptionalStringCollectionAsList();
-            this.connectorNames = in.readOptionalStringCollectionAsList();
-            this.connectorServiceTypes = in.readOptionalStringCollectionAsList();
-            this.connectorSearchQuery = in.readOptionalString();
-        }
 
         public Request(
             PageParams pageParams,
             List<String> indexNames,
             List<String> connectorNames,
             List<String> serviceTypes,
-            String connectorSearchQuery
+            String connectorSearchQuery,
+            Boolean includeDeleted
         ) {
             this.pageParams = pageParams;
             this.indexNames = indexNames;
             this.connectorNames = connectorNames;
             this.connectorServiceTypes = serviceTypes;
             this.connectorSearchQuery = connectorSearchQuery;
+            this.includeDeleted = includeDeleted;
         }
 
         public PageParams getPageParams() {
@@ -95,6 +91,10 @@ public class ListConnectorAction {
             return connectorSearchQuery;
         }
 
+        public Boolean getIncludeDeleted() {
+            return includeDeleted;
+        }
+
         @Override
         public ActionRequestValidationException validate() {
             ActionRequestValidationException validationException = null;
@@ -114,12 +114,7 @@ public class ListConnectorAction {
 
         @Override
         public void writeTo(StreamOutput out) throws IOException {
-            super.writeTo(out);
+            TransportAction.localOnly();
-            pageParams.writeTo(out);
-            out.writeOptionalStringCollection(indexNames);
-            out.writeOptionalStringCollection(connectorNames);
-            out.writeOptionalStringCollection(connectorServiceTypes);
-            out.writeOptionalString(connectorSearchQuery);
         }
 
         @Override
@@ -131,7 +126,8 @@ public class ListConnectorAction {
                 && Objects.equals(indexNames, request.indexNames)
                 && Objects.equals(connectorNames, request.connectorNames)
                 && Objects.equals(connectorServiceTypes, request.connectorServiceTypes)
-                && Objects.equals(connectorSearchQuery, request.connectorSearchQuery);
+                && Objects.equals(connectorSearchQuery, request.connectorSearchQuery)
+                && Objects.equals(includeDeleted, request.includeDeleted);
         }
 
         @Override
@@ -147,7 +143,8 @@ public class ListConnectorAction {
                 (List<String>) p[1],
                 (List<String>) p[2],
                 (List<String>) p[3],
-                (String) p[4]
+                (String) p[4],
+                (Boolean) p[5]
             )
         );
 
@@ -157,6 +154,7 @@ public class ListConnectorAction {
             PARSER.declareStringArray(optionalConstructorArg(), NAMES_FIELD);
             PARSER.declareStringArray(optionalConstructorArg(), Connector.SERVICE_TYPE_FIELD);
             PARSER.declareString(optionalConstructorArg(), SEARCH_QUERY_FIELD);
+            PARSER.declareBoolean(optionalConstructorArg(), INCLUDE_DELETED_FIELD);
         }
 
         public static ListConnectorAction.Request parse(XContentParser parser) {
@@ -172,6 +170,7 @@ public class ListConnectorAction {
                 builder.field(NAMES_FIELD.getPreferredName(), connectorNames);
                 builder.field(Connector.SERVICE_TYPE_FIELD.getPreferredName(), connectorServiceTypes);
                 builder.field(SEARCH_QUERY_FIELD.getPreferredName(), connectorSearchQuery);
+                builder.field(INCLUDE_DELETED_FIELD.getPreferredName(), includeDeleted);
             }
             builder.endObject();
             return builder;

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

@@ -36,7 +36,8 @@ public class RestGetConnectorAction extends BaseRestHandler {
 
     @Override
     protected RestChannelConsumer prepareRequest(RestRequest restRequest, NodeClient client) {
-        GetConnectorAction.Request request = new GetConnectorAction.Request(restRequest.param(CONNECTOR_ID_PARAM));
+        Boolean includeDeleted = restRequest.paramAsBoolean("include_deleted", false);
+        GetConnectorAction.Request request = new GetConnectorAction.Request(restRequest.param(CONNECTOR_ID_PARAM), includeDeleted);
         return channel -> client.execute(GetConnectorAction.INSTANCE, request, new RestToXContentListener<>(channel));
     }
 }

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

@@ -43,13 +43,15 @@ public class RestListConnectorAction extends BaseRestHandler {
         List<String> connectorNames = List.of(restRequest.paramAsStringArray("connector_name", new String[0]));
         List<String> serviceTypes = List.of(restRequest.paramAsStringArray("service_type", new String[0]));
         String searchQuery = restRequest.param("query");
+        Boolean includeDeleted = restRequest.paramAsBoolean("include_deleted", false);
 
         ListConnectorAction.Request request = new ListConnectorAction.Request(
             new PageParams(from, size),
             indexNames,
             connectorNames,
             serviceTypes,
-            searchQuery
+            searchQuery,
+            includeDeleted
         );
 
         return channel -> client.execute(ListConnectorAction.INSTANCE, request, new RestToXContentListener<>(channel));

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

@@ -9,7 +9,7 @@ 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.action.support.TransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.injection.guice.Inject;
@@ -17,23 +17,21 @@ import org.elasticsearch.tasks.Task;
 import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.application.connector.ConnectorIndexService;
 
-public class TransportGetConnectorAction extends HandledTransportAction<GetConnectorAction.Request, GetConnectorAction.Response> {
+public class TransportGetConnectorAction extends TransportAction<GetConnectorAction.Request, GetConnectorAction.Response> {
     protected final ConnectorIndexService connectorIndexService;
 
     @Inject
     public TransportGetConnectorAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(
+        super(GetConnectorAction.NAME, actionFilters, transportService.getTaskManager(), EsExecutors.DIRECT_EXECUTOR_SERVICE);
-            GetConnectorAction.NAME,
-            transportService,
-            actionFilters,
-            GetConnectorAction.Request::new,
-            EsExecutors.DIRECT_EXECUTOR_SERVICE
-        );
         this.connectorIndexService = new ConnectorIndexService(client);
     }
 
     @Override
     protected void doExecute(Task task, GetConnectorAction.Request request, ActionListener<GetConnectorAction.Response> listener) {
-        connectorIndexService.getConnector(request.getConnectorId(), listener.map(GetConnectorAction.Response::new));
+        connectorIndexService.getConnector(
+            request.getConnectorId(),
+            request.getIncludeDeleted(),
+            listener.map(GetConnectorAction.Response::new)
+        );
     }
 }

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

@@ -9,7 +9,7 @@ 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.action.support.TransportAction;
 import org.elasticsearch.client.internal.Client;
 import org.elasticsearch.common.util.concurrent.EsExecutors;
 import org.elasticsearch.injection.guice.Inject;
@@ -18,18 +18,12 @@ import org.elasticsearch.transport.TransportService;
 import org.elasticsearch.xpack.application.connector.ConnectorIndexService;
 import org.elasticsearch.xpack.core.action.util.PageParams;
 
-public class TransportListConnectorAction extends HandledTransportAction<ListConnectorAction.Request, ListConnectorAction.Response> {
+public class TransportListConnectorAction extends TransportAction<ListConnectorAction.Request, ListConnectorAction.Response> {
     protected final ConnectorIndexService connectorIndexService;
 
     @Inject
     public TransportListConnectorAction(TransportService transportService, ActionFilters actionFilters, Client client) {
-        super(
+        super(ListConnectorAction.NAME, actionFilters, transportService.getTaskManager(), EsExecutors.DIRECT_EXECUTOR_SERVICE);
-            ListConnectorAction.NAME,
-            transportService,
-            actionFilters,
-            ListConnectorAction.Request::new,
-            EsExecutors.DIRECT_EXECUTOR_SERVICE
-        );
         this.connectorIndexService = new ConnectorIndexService(client);
     }
 
@@ -44,6 +38,8 @@ public class TransportListConnectorAction extends HandledTransportAction<ListCon
             request.getConnectorNames(),
             request.getConnectorServiceTypes(),
             request.getConnectorSearchQuery(),
+            request.getIncludeDeleted(),
+
             listener.map(r -> new ListConnectorAction.Response(r.connectors(), r.totalResults()))
         );
     }

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

@@ -10,9 +10,6 @@ package org.elasticsearch.xpack.application.connector.syncjob;
 import org.elasticsearch.ElasticsearchParseException;
 import org.elasticsearch.common.Strings;
 import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.io.stream.StreamInput;
-import org.elasticsearch.common.io.stream.StreamOutput;
-import org.elasticsearch.common.io.stream.Writeable;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.core.Nullable;
 import org.elasticsearch.xcontent.ConstructingObjectParser;
@@ -65,7 +62,7 @@ import static org.elasticsearch.xcontent.ConstructingObjectParser.optionalConstr
  *     <li>The hostname of the worker to run the sync job.</li>
  * </ul>
  */
-public class ConnectorSyncJob implements Writeable, ToXContentObject {
+public class ConnectorSyncJob implements ToXContentObject {
 
     static final ParseField CANCELATION_REQUESTED_AT_FIELD = new ParseField("cancelation_requested_at");
 
@@ -213,27 +210,6 @@ public class ConnectorSyncJob implements Writeable, ToXContentObject {
         this.workerHostname = workerHostname;
     }
 
-    public ConnectorSyncJob(StreamInput in) throws IOException {
-        this.cancelationRequestedAt = in.readOptionalInstant();
-        this.canceledAt = in.readOptionalInstant();
-        this.completedAt = in.readOptionalInstant();
-        this.connector = in.readNamedWriteable(Connector.class);
-        this.createdAt = in.readInstant();
-        this.deletedDocumentCount = in.readLong();
-        this.error = in.readOptionalString();
-        this.id = in.readString();
-        this.indexedDocumentCount = in.readLong();
-        this.indexedDocumentVolume = in.readLong();
-        this.jobType = in.readEnum(ConnectorSyncJobType.class);
-        this.lastSeen = in.readOptionalInstant();
-        this.metadata = in.readMap(StreamInput::readString, StreamInput::readGenericValue);
-        this.startedAt = in.readOptionalInstant();
-        this.status = in.readEnum(ConnectorSyncStatus.class);
-        this.totalDocumentCount = in.readOptionalLong();
-        this.triggerMethod = in.readEnum(ConnectorSyncJobTriggerMethod.class);
-        this.workerHostname = in.readOptionalString();
-    }
-
     @SuppressWarnings("unchecked")
     private static final ConstructingObjectParser<ConnectorSyncJob, String> PARSER = new ConstructingObjectParser<>(
         "connector_sync_job",
@@ -548,28 +524,6 @@ public class ConnectorSyncJob implements Writeable, ToXContentObject {
         return builder;
     }
 
-    @Override
-    public void writeTo(StreamOutput out) throws IOException {
-        out.writeOptionalInstant(cancelationRequestedAt);
-        out.writeOptionalInstant(canceledAt);
-        out.writeOptionalInstant(completedAt);
-        out.writeNamedWriteable(connector);
-        out.writeInstant(createdAt);
-        out.writeLong(deletedDocumentCount);
-        out.writeOptionalString(error);
-        out.writeString(id);
-        out.writeLong(indexedDocumentCount);
-        out.writeLong(indexedDocumentVolume);
-        out.writeEnum(jobType);
-        out.writeOptionalInstant(lastSeen);
-        out.writeMap(metadata, StreamOutput::writeString, StreamOutput::writeGenericValue);
-        out.writeOptionalInstant(startedAt);
-        out.writeEnum(status);
-        out.writeOptionalLong(totalDocumentCount);
-        out.writeEnum(triggerMethod);
-        out.writeOptionalString(workerHostname);
-    }
-
     public boolean equals(Object other) {
         if (this == other) return true;
         if (other == null || getClass() != other.getClass()) return false;

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

@@ -11,7 +11,6 @@ import org.elasticsearch.ElasticsearchStatusException;
 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.common.settings.Settings;
 import org.elasticsearch.core.Tuple;
@@ -112,13 +111,66 @@ public class ConnectorIndexServiceTests extends ESSingleNodeTestCase {
         }
 
         String connectorIdToDelete = connectorIds.get(0);
-        DeleteResponse resp = awaitDeleteConnector(connectorIdToDelete, false);
+        UpdateResponse resp = awaitDeleteConnector(connectorIdToDelete, false);
         assertThat(resp.status(), equalTo(RestStatus.OK));
         expectThrows(ResourceNotFoundException.class, () -> awaitGetConnector(connectorIdToDelete));
 
         expectThrows(ResourceNotFoundException.class, () -> awaitDeleteConnector(connectorIdToDelete, false));
     }
 
+    public void testDeleteConnector_expectSoftDeletion() throws Exception {
+        int numConnectors = 5;
+        List<String> connectorIds = new ArrayList<>();
+        List<Connector> connectors = new ArrayList<>();
+        for (int i = 0; i < numConnectors; i++) {
+            Connector connector = ConnectorTestUtils.getRandomConnector();
+            ConnectorCreateActionResponse resp = awaitCreateConnector(null, connector);
+            connectorIds.add(resp.getId());
+            connectors.add(connector);
+        }
+
+        String connectorIdToDelete = connectorIds.get(0);
+        UpdateResponse resp = awaitDeleteConnector(connectorIdToDelete, false);
+        assertThat(resp.status(), equalTo(RestStatus.OK));
+        expectThrows(ResourceNotFoundException.class, () -> awaitGetConnector(connectorIdToDelete));
+
+        expectThrows(ResourceNotFoundException.class, () -> awaitDeleteConnector(connectorIdToDelete, false));
+
+        Connector softDeletedConnector = awaitGetSoftDeletedConnector(connectorIdToDelete);
+        assertThat(softDeletedConnector.getConnectorId(), equalTo(connectorIdToDelete));
+        assertThat(softDeletedConnector.getServiceType(), equalTo(connectors.get(0).getServiceType()));
+    }
+
+    public void testDeleteConnector_expectSoftDeletionMultipleConnectors() throws Exception {
+        int numConnectors = 5;
+        List<String> connectorIds = new ArrayList<>();
+        for (int i = 0; i < numConnectors; i++) {
+            Connector connector = ConnectorTestUtils.getRandomConnector();
+            ConnectorCreateActionResponse resp = awaitCreateConnector(null, connector);
+            connectorIds.add(resp.getId());
+        }
+
+        // Delete all of them
+        for (int i = 0; i < numConnectors; i++) {
+            String connectorIdToDelete = connectorIds.get(i);
+            UpdateResponse resp = awaitDeleteConnector(connectorIdToDelete, false);
+            assertThat(resp.status(), equalTo(RestStatus.OK));
+        }
+
+        // Connectors were deleted from main index
+        for (int i = 0; i < numConnectors; i++) {
+            String connectorId = connectorIds.get(i);
+            expectThrows(ResourceNotFoundException.class, () -> awaitGetConnector(connectorId));
+        }
+
+        // Soft deleted connectors available in system index
+        for (int i = 0; i < numConnectors; i++) {
+            String connectorId = connectorIds.get(i);
+            Connector softDeletedConnector = awaitGetSoftDeletedConnector(connectorId);
+            assertThat(softDeletedConnector.getConnectorId(), equalTo(connectorId));
+        }
+    }
+
     public void testUpdateConnectorConfiguration_FullConfiguration() throws Exception {
         Connector connector = ConnectorTestUtils.getRandomConnector();
         String connectorId = randomUUID();
@@ -897,13 +949,13 @@ public class ConnectorIndexServiceTests extends ESSingleNodeTestCase {
         assertThat(updateApiKeyIdRequest.getApiKeySecretId(), equalTo(indexedConnector.getApiKeySecretId()));
     }
 
-    private DeleteResponse awaitDeleteConnector(String connectorId, boolean deleteConnectorSyncJobs) throws Exception {
+    private UpdateResponse awaitDeleteConnector(String connectorId, boolean deleteConnectorSyncJobs) throws Exception {
         CountDownLatch latch = new CountDownLatch(1);
-        final AtomicReference<DeleteResponse> resp = new AtomicReference<>(null);
+        final AtomicReference<UpdateResponse> resp = new AtomicReference<>(null);
         final AtomicReference<Exception> exc = new AtomicReference<>(null);
         connectorIndexService.deleteConnector(connectorId, deleteConnectorSyncJobs, new ActionListener<>() {
             @Override
-            public void onResponse(DeleteResponse deleteResponse) {
+            public void onResponse(UpdateResponse deleteResponse) {
                 resp.set(deleteResponse);
                 latch.countDown();
             }
@@ -956,11 +1008,19 @@ public class ConnectorIndexServiceTests extends ESSingleNodeTestCase {
         return resp.get();
     }
 
+    private Connector awaitGetSoftDeletedConnector(String connectorId) throws Exception {
+        return awaitGetConnector(connectorId, true);
+    }
+
     private Connector awaitGetConnector(String connectorId) throws Exception {
+        return awaitGetConnector(connectorId, false);
+    }
+
+    private Connector awaitGetConnector(String connectorId, Boolean isDeleted) throws Exception {
         CountDownLatch latch = new CountDownLatch(1);
         final AtomicReference<Connector> resp = new AtomicReference<>(null);
         final AtomicReference<Exception> exc = new AtomicReference<>(null);
-        connectorIndexService.getConnector(connectorId, new ActionListener<>() {
+        connectorIndexService.getConnector(connectorId, isDeleted, new ActionListener<>() {
             @Override
             public void onResponse(ConnectorSearchResult connectorResult) {
                 // Serialize the sourceRef to Connector class for unit tests
@@ -994,11 +1054,23 @@ public class ConnectorIndexServiceTests extends ESSingleNodeTestCase {
         List<String> names,
         List<String> serviceTypes,
         String searchQuery
+    ) throws Exception {
+        return awaitListConnector(from, size, indexNames, names, serviceTypes, searchQuery, false);
+    }
+
+    private ConnectorIndexService.ConnectorResult awaitListConnector(
+        int from,
+        int size,
+        List<String> indexNames,
+        List<String> names,
+        List<String> serviceTypes,
+        String searchQuery,
+        Boolean isDeleted
     ) throws Exception {
         CountDownLatch latch = new CountDownLatch(1);
         final AtomicReference<ConnectorIndexService.ConnectorResult> resp = new AtomicReference<>(null);
         final AtomicReference<Exception> exc = new AtomicReference<>(null);
-        connectorIndexService.listConnectors(from, size, indexNames, names, serviceTypes, searchQuery, new ActionListener<>() {
+        connectorIndexService.listConnectors(from, size, indexNames, names, serviceTypes, searchQuery, isDeleted, new ActionListener<>() {
             @Override
             public void onResponse(ConnectorIndexService.ConnectorResult result) {
                 resp.set(result);

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

@@ -9,43 +9,20 @@ package org.elasticsearch.xpack.application.connector;
 
 import org.elasticsearch.common.bytes.BytesArray;
 import org.elasticsearch.common.bytes.BytesReference;
-import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
-import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.xcontent.XContentHelper;
-import org.elasticsearch.search.SearchModule;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.ToXContent;
 import org.elasticsearch.xcontent.XContentParser;
 import org.elasticsearch.xcontent.XContentType;
-import org.junit.Before;
 
 import java.io.IOException;
-import java.util.List;
 
-import static java.util.Collections.emptyList;
 import static org.elasticsearch.common.xcontent.XContentHelper.toXContent;
 import static org.elasticsearch.test.hamcrest.ElasticsearchAssertions.assertToXContentEquivalent;
 import static org.hamcrest.CoreMatchers.equalTo;
 
 public class ConnectorTests extends ESTestCase {
 
-    private NamedWriteableRegistry namedWriteableRegistry;
-
-    @Before
-    public void registerNamedObjects() {
-        SearchModule searchModule = new SearchModule(Settings.EMPTY, emptyList());
-
-        List<NamedWriteableRegistry.Entry> namedWriteables = searchModule.getNamedWriteables();
-        namedWriteableRegistry = new NamedWriteableRegistry(namedWriteables);
-    }
-
-    public final void testRandomSerialization() throws IOException {
-        for (int runs = 0; runs < 10; runs++) {
-            Connector testInstance = ConnectorTestUtils.getRandomConnector();
-            assertTransportSerialization(testInstance);
-        }
-    }
-
     public void testToXContent() throws IOException {
         String connectorId = "test-connector";
         String content = XContentHelper.stripWhitespace("""
@@ -383,14 +360,4 @@ public class ConnectorTests extends ESTestCase {
         assertToXContentEquivalent(originalBytes, toXContent(parsed, XContentType.JSON, humanReadable), XContentType.JSON);
         assertThat(parsed.getApiKeySecretId(), equalTo(null));
     }
-
-    private void assertTransportSerialization(Connector testInstance) throws IOException {
-        Connector deserializedInstance = copyInstance(testInstance);
-        assertNotSame(testInstance, deserializedInstance);
-        assertThat(testInstance, equalTo(deserializedInstance));
-    }
-
-    private Connector copyInstance(Connector instance) throws IOException {
-        return copyWriteable(instance, namedWriteableRegistry, Connector::new);
-    }
 }

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

@@ -1,43 +0,0 @@
-/*
- * 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.test.AbstractBWCSerializationTestCase;
-import org.elasticsearch.xcontent.XContentParser;
-
-import java.io.IOException;
-
-public class GetConnectorActionRequestBWCSerializingTests extends AbstractBWCSerializationTestCase<GetConnectorAction.Request> {
-
-    @Override
-    protected Writeable.Reader<GetConnectorAction.Request> instanceReader() {
-        return GetConnectorAction.Request::new;
-    }
-
-    @Override
-    protected GetConnectorAction.Request createTestInstance() {
-        return new GetConnectorAction.Request(randomAlphaOfLengthBetween(1, 10));
-    }
-
-    @Override
-    protected GetConnectorAction.Request mutateInstance(GetConnectorAction.Request instance) throws IOException {
-        return randomValueOtherThan(instance, this::createTestInstance);
-    }
-
-    @Override
-    protected GetConnectorAction.Request doParseInstance(XContentParser parser) throws IOException {
-        return GetConnectorAction.Request.parse(parser);
-    }
-
-    @Override
-    protected GetConnectorAction.Request mutateInstanceForVersion(GetConnectorAction.Request instance, TransportVersion version) {
-        return new GetConnectorAction.Request(instance.getConnectorId());
-    }
-}

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

@@ -1,58 +0,0 @@
-/*
- * 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.test.AbstractBWCSerializationTestCase;
-import org.elasticsearch.xcontent.XContentParser;
-import org.elasticsearch.xpack.application.EnterpriseSearchModuleTestUtils;
-import org.elasticsearch.xpack.core.action.util.PageParams;
-
-import java.io.IOException;
-import java.util.List;
-
-public class ListConnectorActionRequestBWCSerializingTests extends AbstractBWCSerializationTestCase<ListConnectorAction.Request> {
-    @Override
-    protected Writeable.Reader<ListConnectorAction.Request> instanceReader() {
-        return ListConnectorAction.Request::new;
-    }
-
-    @Override
-    protected ListConnectorAction.Request createTestInstance() {
-        PageParams pageParams = EnterpriseSearchModuleTestUtils.randomPageParams();
-        return new ListConnectorAction.Request(
-            pageParams,
-            List.of(generateRandomStringArray(10, 10, false)),
-            List.of(generateRandomStringArray(10, 10, false)),
-            List.of(generateRandomStringArray(10, 10, false)),
-            randomAlphaOfLengthBetween(3, 10)
-        );
-    }
-
-    @Override
-    protected ListConnectorAction.Request mutateInstance(ListConnectorAction.Request instance) throws IOException {
-        return randomValueOtherThan(instance, this::createTestInstance);
-    }
-
-    @Override
-    protected ListConnectorAction.Request doParseInstance(XContentParser parser) throws IOException {
-        return ListConnectorAction.Request.parse(parser);
-    }
-
-    @Override
-    protected ListConnectorAction.Request mutateInstanceForVersion(ListConnectorAction.Request instance, TransportVersion version) {
-        return new ListConnectorAction.Request(
-            instance.getPageParams(),
-            instance.getIndexNames(),
-            instance.getConnectorNames(),
-            instance.getConnectorServiceTypes(),
-            instance.getConnectorSearchQuery()
-        );
-    }
-}

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

@@ -8,22 +8,14 @@
 package org.elasticsearch.xpack.application.connector.secrets.action;
 
 import org.elasticsearch.TransportVersion;
-import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.xpack.application.connector.Connector;
 import org.elasticsearch.xpack.application.connector.secrets.ConnectorSecretsTestUtils;
 import org.elasticsearch.xpack.core.ml.AbstractBWCWireSerializationTestCase;
 
 import java.io.IOException;
-import java.util.List;
 
 public class DeleteConnectorSecretResponseBWCSerializingTests extends AbstractBWCWireSerializationTestCase<DeleteConnectorSecretResponse> {
 
-    @Override
-    public NamedWriteableRegistry getNamedWriteableRegistry() {
-        return new NamedWriteableRegistry(List.of(new NamedWriteableRegistry.Entry(Connector.class, Connector.NAME, Connector::new)));
-    }
-
     @Override
     protected Writeable.Reader<DeleteConnectorSecretResponse> instanceReader() {
         return DeleteConnectorSecretResponse::new;

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

@@ -8,22 +8,14 @@
 package org.elasticsearch.xpack.application.connector.secrets.action;
 
 import org.elasticsearch.TransportVersion;
-import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.xpack.application.connector.Connector;
 import org.elasticsearch.xpack.application.connector.secrets.ConnectorSecretsTestUtils;
 import org.elasticsearch.xpack.core.ml.AbstractBWCWireSerializationTestCase;
 
 import java.io.IOException;
-import java.util.List;
 
 public class GetConnectorSecretResponseBWCSerializingTests extends AbstractBWCWireSerializationTestCase<GetConnectorSecretResponse> {
 
-    @Override
-    public NamedWriteableRegistry getNamedWriteableRegistry() {
-        return new NamedWriteableRegistry(List.of(new NamedWriteableRegistry.Entry(Connector.class, Connector.NAME, Connector::new)));
-    }
-
     @Override
     protected Writeable.Reader<GetConnectorSecretResponse> instanceReader() {
         return GetConnectorSecretResponse::new;

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

@@ -8,17 +8,14 @@
 package org.elasticsearch.xpack.application.connector.syncjob;
 
 import org.elasticsearch.common.bytes.BytesArray;
-import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.test.ESTestCase;
 import org.elasticsearch.xcontent.XContentType;
 import org.elasticsearch.xpack.application.connector.Connector;
 import org.elasticsearch.xpack.application.connector.ConnectorSyncStatus;
-import org.junit.Before;
 
 import java.io.IOException;
 import java.time.Instant;
-import java.util.List;
 
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.hasSize;
@@ -26,22 +23,6 @@ import static org.hamcrest.Matchers.notNullValue;
 
 public class ConnectorSyncJobTests extends ESTestCase {
 
-    private NamedWriteableRegistry namedWriteableRegistry;
-
-    @Before
-    public void registerNamedObjects() {
-        namedWriteableRegistry = new NamedWriteableRegistry(
-            List.of(new NamedWriteableRegistry.Entry(Connector.class, Connector.NAME, Connector::new))
-        );
-    }
-
-    public final void testRandomSerialization() throws IOException {
-        for (int run = 0; run < 10; run++) {
-            ConnectorSyncJob syncJob = ConnectorSyncJobTestUtils.getRandomConnectorSyncJob();
-            assertTransportSerialization(syncJob);
-        }
-    }
-
     public void testFromXContent_WithAllFields_AllSet() throws IOException {
         String content = XContentHelper.stripWhitespace("""
             {
@@ -332,14 +313,4 @@ public class ConnectorSyncJobTests extends ESTestCase {
 
         ConnectorSyncJob.syncJobConnectorFromXContentBytes(new BytesArray(content), null, XContentType.JSON);
     }
-
-    private void assertTransportSerialization(ConnectorSyncJob testInstance) throws IOException {
-        ConnectorSyncJob deserializedInstance = copyInstance(testInstance);
-        assertNotSame(testInstance, deserializedInstance);
-        assertThat(testInstance, equalTo(deserializedInstance));
-    }
-
-    private ConnectorSyncJob copyInstance(ConnectorSyncJob instance) throws IOException {
-        return copyWriteable(instance, namedWriteableRegistry, ConnectorSyncJob::new);
-    }
 }

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

@@ -8,23 +8,15 @@
 package org.elasticsearch.xpack.application.connector.syncjob.action;
 
 import org.elasticsearch.TransportVersion;
-import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.xpack.application.connector.Connector;
 import org.elasticsearch.xpack.application.connector.syncjob.ConnectorSyncJobTestUtils;
 import org.elasticsearch.xpack.core.ml.AbstractBWCWireSerializationTestCase;
 
 import java.io.IOException;
-import java.util.List;
 
 public class GetConnectorSyncJobActionResponseBWCSerializingTests extends AbstractBWCWireSerializationTestCase<
     GetConnectorSyncJobAction.Response> {
 
-    @Override
-    public NamedWriteableRegistry getNamedWriteableRegistry() {
-        return new NamedWriteableRegistry(List.of(new NamedWriteableRegistry.Entry(Connector.class, Connector.NAME, Connector::new)));
-    }
-
     @Override
     protected Writeable.Reader<GetConnectorSyncJobAction.Response> instanceReader() {
         return GetConnectorSyncJobAction.Response::new;

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

@@ -8,23 +8,15 @@
 package org.elasticsearch.xpack.application.connector.syncjob.action;
 
 import org.elasticsearch.TransportVersion;
-import org.elasticsearch.common.io.stream.NamedWriteableRegistry;
 import org.elasticsearch.common.io.stream.Writeable;
-import org.elasticsearch.xpack.application.connector.Connector;
 import org.elasticsearch.xpack.application.connector.syncjob.ConnectorSyncJobTestUtils;
 import org.elasticsearch.xpack.core.ml.AbstractBWCWireSerializationTestCase;
 
 import java.io.IOException;
-import java.util.List;
 
 public class ListConnectorSyncJobsActionResponseBWCSerializingTests extends AbstractBWCWireSerializationTestCase<
     ListConnectorSyncJobsAction.Response> {
 
-    @Override
-    protected NamedWriteableRegistry getNamedWriteableRegistry() {
-        return new NamedWriteableRegistry(List.of(new NamedWriteableRegistry.Entry(Connector.class, Connector.NAME, Connector::new)));
-    }
-
     @Override
     protected Writeable.Reader<ListConnectorSyncJobsAction.Response> instanceReader() {
         return ListConnectorSyncJobsAction.Response::new;