applier

Author: ywangd

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3PerProjectClientManager.java

@@ -12,7 +12,7 @@
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.support.SubscribableListener;
 import org.elasticsearch.cluster.ClusterChangedEvent;
-import org.elasticsearch.cluster.ClusterStateListener;
+import org.elasticsearch.cluster.ClusterStateApplier;
 import org.elasticsearch.cluster.metadata.ProjectId;
 import org.elasticsearch.cluster.metadata.ProjectMetadata;
 import org.elasticsearch.common.settings.ProjectSecrets;
@@ -37,7 +37,7 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
-public class S3PerProjectClientManager implements ClusterStateListener {
+public class S3PerProjectClientManager implements ClusterStateApplier {
 
     private static final Logger logger = LogManager.getLogger(S3PerProjectClientManager.class);
     private static final String S3_SETTING_PREFIX = "s3.";
@@ -65,7 +65,8 @@ Map<ProjectId, ClientsHolder> getProjectClientsHolders() {
         return Map.copyOf(projectClientsHolders);
     }
 
-    public void clusterChanged(ClusterChangedEvent event) {
+    @Override
+    public void applyClusterState(ClusterChangedEvent event) {
         final Map<ProjectId, ProjectMetadata> currentProjects = event.state().metadata().projects();
 
         final var updatedPerProjectClients = new HashMap<ProjectId, ClientsHolder>();

modules/repository-s3/src/main/java/org/elasticsearch/repositories/s3/S3Service.java

@@ -155,7 +155,7 @@ class S3Service extends AbstractLifecycleComponent {
                 this::buildClientReference,
                 clusterService.threadPool().generic()
             );
-            clusterService.addListener(s3PerProjectClientManager);
+            clusterService.addHighPriorityApplier(s3PerProjectClientManager);
         } else {
             s3PerProjectClientManager = null;
         }

modules/repository-s3/src/test/java/org/elasticsearch/repositories/s3/S3ServiceTests.java

@@ -319,10 +319,11 @@ public void testEndpointOverrideSchemeUsesHttpIfHttpProtocolSpecified() {
             See the breaking changes documentation for the next major version.""", clientName));
     }
 
-    private static URI getEndpointUri(Settings.Builder settings, String clientName) {
+    private URI getEndpointUri(Settings.Builder settings, String clientName) {
         return new S3Service(
             mock(Environment.class),
-            Settings.EMPTY,
+            ClusterServiceUtils.createClusterService(threadPool),
+            TestProjectResolvers.DEFAULT_PROJECT_ONLY,
             mock(ResourceWatcherService.class),
             () -> Region.of(randomIdentifier())
         ).buildClient(S3ClientSettings.getClientSettings(settings.build(), clientName), mock(SdkHttpClient.class))