[Connector API] Fix bug with with wrong target index for access control sync (#109097) (#109157)

Author: jedrazb

docs/changelog/109097.yaml

@@ -0,0 +1,6 @@
+pr: 109097
+summary: "[Connector API] Fix bug with with wrong target index for access control\
+  \ sync"
+area: Application
+type: bug
+issues: []

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

@@ -266,6 +266,27 @@ setup:
   - exists: created_at
   - exists: last_seen
 
+---
+'Create access control sync job - expect prefixed connector index name':
+  - do:
+      connector.sync_job_post:
+        body:
+          id: test-connector
+          job_type: access_control
+
+  - set:  { id: id }
+
+  - match: { id: $id }
+
+  - do:
+      connector.sync_job_get:
+        connector_sync_job_id: $id
+
+  - match: { connector.id: test-connector }
+  - match: { job_type: access_control }
+  - match: { connector.index_name: .search-acl-filter-search-test }
+
+
 ---
 'Create connector sync job with non-existing connector id':
   - do:

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

@@ -34,6 +34,7 @@
 import java.util.Objects;
 
 import static org.elasticsearch.xcontent.ConstructingObjectParser.optionalConstructorArg;
+import static org.elasticsearch.xpack.application.connector.ConnectorTemplateRegistry.ACCESS_CONTROL_INDEX_PREFIX;
 
 /**
  * Represents a Connector in the Elasticsearch ecosystem. Connectors are used for integrating
@@ -269,6 +270,10 @@ public Connector(StreamInput in) throws IOException {
         }
     );
 
+    public String getAccessControlIndexName() {
+        return ACCESS_CONTROL_INDEX_PREFIX + this.indexName;
+    }
+
     static {
         PARSER.declareStringOrNull(optionalConstructorArg(), API_KEY_ID_FIELD);
         PARSER.declareStringOrNull(optionalConstructorArg(), API_KEY_SECRET_ID_FIELD);

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

@@ -46,6 +46,7 @@ public class ConnectorTemplateRegistry extends IndexTemplateRegistry {
     public static final String CONNECTOR_SYNC_JOBS_INDEX_NAME_PATTERN = ".elastic-connectors-sync-jobs-v1";
     public static final String CONNECTOR_SYNC_JOBS_TEMPLATE_NAME = "elastic-connectors-sync-jobs";
 
+    public static final String ACCESS_CONTROL_INDEX_PREFIX = ".search-acl-filter-";
     public static final String ACCESS_CONTROL_INDEX_NAME_PATTERN = ".search-acl-filter-*";
     public static final String ACCESS_CONTROL_TEMPLATE_NAME = "search-acl-filter";
 

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

@@ -96,9 +96,9 @@ public void createConnectorSyncJob(
         ActionListener<PostConnectorSyncJobAction.Response> listener
     ) {
         String connectorId = request.getId();
+        ConnectorSyncJobType jobType = Objects.requireNonNullElse(request.getJobType(), ConnectorSyncJob.DEFAULT_JOB_TYPE);
         try {
-            getSyncJobConnectorInfo(connectorId, listener.delegateFailure((l, connector) -> {
-
+            getSyncJobConnectorInfo(connectorId, jobType, listener.delegateFailure((l, connector) -> {
                 if (Strings.isNullOrEmpty(connector.getIndexName())) {
                     l.onFailure(
                         new ElasticsearchStatusException(
@@ -113,7 +113,6 @@ public void createConnectorSyncJob(
                 }
 
                 Instant now = Instant.now();
-                ConnectorSyncJobType jobType = Objects.requireNonNullElse(request.getJobType(), ConnectorSyncJob.DEFAULT_JOB_TYPE);
                 ConnectorSyncJobTriggerMethod triggerMethod = Objects.requireNonNullElse(
                     request.getTriggerMethod(),
                     ConnectorSyncJob.DEFAULT_TRIGGER_METHOD
@@ -482,7 +481,7 @@ private ConnectorSyncStatus getConnectorSyncJobStatusFromSearchResult(ConnectorS
         );
     }
 
-    private void getSyncJobConnectorInfo(String connectorId, ActionListener<Connector> listener) {
+    private void getSyncJobConnectorInfo(String connectorId, ConnectorSyncJobType jobType, ActionListener<Connector> listener) {
         try {
 
             final GetRequest request = new GetRequest(ConnectorIndexService.CONNECTOR_INDEX_NAME, connectorId);
@@ -502,11 +501,15 @@ public void onResponse(GetResponse response) {
                             connectorId,
                             XContentType.JSON
                         );
+                        // Access control syncs write data to a separate index
+                        String targetIndexName = jobType == ConnectorSyncJobType.ACCESS_CONTROL
+                            ? connector.getAccessControlIndexName()
+                            : connector.getIndexName();
 
                         // Build the connector representation for sync job
                         final Connector syncJobConnector = new Connector.Builder().setConnectorId(connector.getConnectorId())
                             .setSyncJobFiltering(transformConnectorFilteringToSyncJobRepresentation(connector.getFiltering()))
-                            .setIndexName(connector.getIndexName())
+                            .setIndexName(targetIndexName)
                             .setLanguage(connector.getLanguage())
                             .setPipeline(connector.getPipeline())
                             .setServiceType(connector.getServiceType())

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

@@ -56,6 +56,7 @@
 import java.util.stream.Collectors;
 
 import static org.elasticsearch.xcontent.XContentFactory.jsonBuilder;
+import static org.elasticsearch.xpack.application.connector.ConnectorTemplateRegistry.ACCESS_CONTROL_INDEX_PREFIX;
 import static org.elasticsearch.xpack.application.connector.ConnectorTestUtils.registerSimplifiedConnectorIndexTemplates;
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
@@ -129,6 +130,18 @@ public void testCreateConnectorSyncJob() throws Exception {
         assertThat(connectorSyncJob.getDeletedDocumentCount(), equalTo(0L));
     }
 
+    public void testCreateConnectorSyncJob_WithAccessControlJobType_IndexIsPrefixed() throws Exception {
+        PostConnectorSyncJobAction.Request createAccessControlJobRequest = ConnectorSyncJobTestUtils
+            .getRandomPostConnectorSyncJobActionRequest(connectorOneId, ConnectorSyncJobType.ACCESS_CONTROL);
+
+        PostConnectorSyncJobAction.Response createAccessControlJobResponse = awaitPutConnectorSyncJob(createAccessControlJobRequest);
+
+        ConnectorSyncJob connectorSyncJob = awaitGetConnectorSyncJob(createAccessControlJobResponse.getId());
+
+        assertThat(connectorSyncJob.getJobType(), equalTo(ConnectorSyncJobType.ACCESS_CONTROL));
+        assertTrue(connectorSyncJob.getConnector().getIndexName().startsWith(ACCESS_CONTROL_INDEX_PREFIX));
+    }
+
     public void testCreateConnectorSyncJob_WithMissingJobType_ExpectDefaultJobTypeToBeSet() throws Exception {
         PostConnectorSyncJobAction.Request syncJobRequest = new PostConnectorSyncJobAction.Request(
             connectorOneId,