sashapolo commented on code in PR #2066:
URL: https://github.com/apache/ignite-3/pull/2066#discussion_r1196143551
##########
modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/ClusterManagementGroupManager.java:
##########
@@ -389,45 +388,44 @@ private void onElectedAsLeader(long term) {
}
});
- raftServiceAfterJoin().thenCompose(this::pushClusterConfigToCluster);
- }
-
- private CompletableFuture<Void> pushClusterConfigToCluster(CmgRaftService
service) {
- return service.readClusterState()
- .thenCompose(state -> {
- if (state == null) {
- LOG.info("No CMG state found in the Raft service");
- return completedFuture(null);
- } else if (state.clusterConfigurationToApply() == null) {
- // Config was applied or wasn't provided
- LOG.info("No cluster configuration found in the Raft
service");
- return completedFuture(null);
- } else {
- LOG.info("Cluster configuration is found in the Raft
service, going to apply it");
- return
distributedConfigurationUpdater.updateConfiguration(state.clusterConfigurationToApply())
- .thenCompose(unused ->
removeClusterConfigFromClusterState(service));
- }
- });
+ raftServiceAfterJoin().whenComplete((service, e) -> {
+ if (e != null) {
+ LOG.error("Error when joining to the raft service", e);
Review Comment:
`joining to the raft service` ? What does that mean?
##########
modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/ClusterManagementGroupManager.java:
##########
@@ -389,45 +388,44 @@ private void onElectedAsLeader(long term) {
}
});
- raftServiceAfterJoin().thenCompose(this::pushClusterConfigToCluster);
- }
-
- private CompletableFuture<Void> pushClusterConfigToCluster(CmgRaftService
service) {
- return service.readClusterState()
- .thenCompose(state -> {
- if (state == null) {
- LOG.info("No CMG state found in the Raft service");
- return completedFuture(null);
- } else if (state.clusterConfigurationToApply() == null) {
- // Config was applied or wasn't provided
- LOG.info("No cluster configuration found in the Raft
service");
- return completedFuture(null);
- } else {
- LOG.info("Cluster configuration is found in the Raft
service, going to apply it");
- return
distributedConfigurationUpdater.updateConfiguration(state.clusterConfigurationToApply())
- .thenCompose(unused ->
removeClusterConfigFromClusterState(service));
- }
- });
+ raftServiceAfterJoin().whenComplete((service, e) -> {
+ if (e != null) {
+ LOG.error("Error when joining to the raft service", e);
+
updateDistributedConfigurationActionFuture.completeExceptionally(e);
+ } else {
+ service.readClusterState()
+ .thenAccept(state -> {
Review Comment:
I'll copy my previous comment here: I think we need to use `whenComplete`
here to complete `updateDistributedConfigurationActionFuture` with exceptions
in case previous operations fail
##########
modules/runner/src/test/java/org/apache/ignite/internal/configuration/DistributedConfigurationUpdaterTest.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.configuration;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.concurrent.CompletableFuture;
+import
org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
+import
org.apache.ignite.internal.cluster.management.UpdateDistributedConfigurationAction;
+import
org.apache.ignite.internal.configuration.presentation.ConfigurationPresentation;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class DistributedConfigurationUpdaterTest {
+
+ @Mock
+ public ConfigurationPresentation<String> presentation;
+
+ @Mock
+ public ClusterManagementGroupManager cmgMgr;
+
+ @Test
+ public void nextActionIsCompletedAfterUpdatingConfiguration() {
+
+ // Set up mocks.
+
when(presentation.update(anyString())).thenReturn(completedFuture(null));
+
+ CompletableFuture<Void> nextAction = new CompletableFuture<>();
+ String configuration = "security.authentication.enabled:true";
+ UpdateDistributedConfigurationAction
updateDistributedConfigurationAction =
+ new UpdateDistributedConfigurationAction(
+ configuration,
+ () -> {
+ nextAction.complete(null);
+ return null;
Review Comment:
Wouldn't this lead to an NPE? Why not return `nextAction`?
##########
modules/runner/src/main/java/org/apache/ignite/internal/configuration/DistributedConfigurationUpdater.java:
##########
@@ -31,37 +31,34 @@ public class DistributedConfigurationUpdater implements
IgniteComponent {
private static final IgniteLogger LOG =
Loggers.forClass(DistributedConfigurationUpdater.class);
- private final CompletableFuture<ConfigurationPresentation<String>>
clusterCfgPresentation = new CompletableFuture<>();
+ private final ClusterManagementGroupManager cmgMgr;
- public void
setDistributedConfigurationPresentation(ConfigurationPresentation<String>
presentation) {
- clusterCfgPresentation.complete(presentation);
+ private final ConfigurationPresentation<String> presentation;
+
+ public DistributedConfigurationUpdater(ClusterManagementGroupManager
cmgMgr, ConfigurationPresentation<String> presentation) {
+ this.cmgMgr = cmgMgr;
+ this.presentation = presentation;
}
- /**
- * Applies changes to the cluster configuration when {@link
DistributedConfigurationUpdater#clusterCfgPresentation}
- * is complete.
- *
- * @param configurationToApply Cluster configuration that should be
applied.
- * @return Future that will be completed when cluster configuration is
updated.
- */
- public CompletableFuture<Void> updateConfiguration(String
configurationToApply) {
- return clusterCfgPresentation.thenCompose(presentation ->
presentation.update(configurationToApply))
+ @Override
+ public void start() {
+ cmgMgr.clusterConfigurationToUpdate()
+ .thenCompose(action -> {
+ if (action.configuration() != null) {
+ return
presentation.update(action.configuration()).thenApply(ignored -> action);
+ } else {
+ return CompletableFuture.completedFuture(action);
+ }
+ })
+ .thenCompose(action -> action.nextAction().get())
Review Comment:
Why do we need to run the `nextAction` even if no configuration was provided
in the previous step?
##########
modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/ClusterManagementGroupManager.java:
##########
@@ -389,45 +388,44 @@ private void onElectedAsLeader(long term) {
}
});
- raftServiceAfterJoin().thenCompose(this::pushClusterConfigToCluster);
- }
-
- private CompletableFuture<Void> pushClusterConfigToCluster(CmgRaftService
service) {
- return service.readClusterState()
- .thenCompose(state -> {
- if (state == null) {
- LOG.info("No CMG state found in the Raft service");
- return completedFuture(null);
- } else if (state.clusterConfigurationToApply() == null) {
- // Config was applied or wasn't provided
- LOG.info("No cluster configuration found in the Raft
service");
- return completedFuture(null);
- } else {
- LOG.info("Cluster configuration is found in the Raft
service, going to apply it");
- return
distributedConfigurationUpdater.updateConfiguration(state.clusterConfigurationToApply())
- .thenCompose(unused ->
removeClusterConfigFromClusterState(service));
- }
- });
+ raftServiceAfterJoin().whenComplete((service, e) -> {
+ if (e != null) {
+ LOG.error("Error when joining to the raft service", e);
+
updateDistributedConfigurationActionFuture.completeExceptionally(e);
+ } else {
+ service.readClusterState()
+ .thenAccept(state -> {
+ String configuration =
state.clusterConfigurationToApply();
+ if (configuration != null) {
+
updateDistributedConfigurationActionFuture.complete(
+ new
UpdateDistributedConfigurationAction(
+ configuration,
+ () ->
removeClusterConfigFromClusterState(service)
+ ));
+ } else {
+
updateDistributedConfigurationActionFuture.cancel(true);
+ }
+ });
+ }
+ });
}
private CompletableFuture<Void>
removeClusterConfigFromClusterState(CmgRaftService service) {
return service.readClusterState()
.thenCompose(state -> {
Review Comment:
I think we should check that configuration is actually present in the
cluster state (i.e. it is not null) and only then update it
##########
modules/runner/src/test/java/org/apache/ignite/internal/configuration/DistributedConfigurationUpdaterTest.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.configuration;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.concurrent.CompletableFuture;
+import
org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
+import
org.apache.ignite.internal.cluster.management.UpdateDistributedConfigurationAction;
+import
org.apache.ignite.internal.configuration.presentation.ConfigurationPresentation;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class DistributedConfigurationUpdaterTest {
+
+ @Mock
+ public ConfigurationPresentation<String> presentation;
+
+ @Mock
+ public ClusterManagementGroupManager cmgMgr;
+
+ @Test
+ public void nextActionIsCompletedAfterUpdatingConfiguration() {
+
+ // Set up mocks.
+
when(presentation.update(anyString())).thenReturn(completedFuture(null));
+
+ CompletableFuture<Void> nextAction = new CompletableFuture<>();
+ String configuration = "security.authentication.enabled:true";
+ UpdateDistributedConfigurationAction
updateDistributedConfigurationAction =
+ new UpdateDistributedConfigurationAction(
+ configuration,
+ () -> {
+ nextAction.complete(null);
+ return null;
+ }
+ );
+
+ when(cmgMgr.clusterConfigurationToUpdate())
+
.thenReturn(completedFuture(updateDistributedConfigurationAction));
+
+ // Run updater.
+ DistributedConfigurationUpdater distributedConfigurationUpdater = new
DistributedConfigurationUpdater(
+ cmgMgr,
+ presentation
+ );
+
+ distributedConfigurationUpdater.start();
+
+ // Verify that configuration was updated.
+ verify(presentation, times(1)).update(configuration);
+
+ // Verify that next action is completed.
+ nextAction.join();
Review Comment:
please use `willCompleteSuccessfully()`
##########
modules/runner/src/test/java/org/apache/ignite/internal/configuration/DistributedConfigurationUpdaterTest.java:
##########
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.configuration;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.concurrent.CompletableFuture;
+import
org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
+import
org.apache.ignite.internal.cluster.management.UpdateDistributedConfigurationAction;
+import
org.apache.ignite.internal.configuration.presentation.ConfigurationPresentation;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+class DistributedConfigurationUpdaterTest {
+
+ @Mock
+ public ConfigurationPresentation<String> presentation;
+
+ @Mock
+ public ClusterManagementGroupManager cmgMgr;
+
+ @Test
+ public void nextActionIsCompletedAfterUpdatingConfiguration() {
+
+ // Set up mocks.
+
when(presentation.update(anyString())).thenReturn(completedFuture(null));
+
+ CompletableFuture<Void> nextAction = new CompletableFuture<>();
+ String configuration = "security.authentication.enabled:true";
+ UpdateDistributedConfigurationAction
updateDistributedConfigurationAction =
+ new UpdateDistributedConfigurationAction(
+ configuration,
+ () -> {
+ nextAction.complete(null);
+ return null;
+ }
+ );
+
+ when(cmgMgr.clusterConfigurationToUpdate())
+
.thenReturn(completedFuture(updateDistributedConfigurationAction));
+
+ // Run updater.
+ DistributedConfigurationUpdater distributedConfigurationUpdater = new
DistributedConfigurationUpdater(
+ cmgMgr,
+ presentation
+ );
+
+ distributedConfigurationUpdater.start();
+
+ // Verify that configuration was updated.
+ verify(presentation, times(1)).update(configuration);
+
+ // Verify that next action is completed.
+ nextAction.join();
+ assertThat(nextAction.isDone(), is(true));
+ }
+
+ @Test
+ public void nextActionIsCompletedIfConfigurationNull() {
+
+ // Set up mocks.
+ CompletableFuture<Void> nextAction = new CompletableFuture<>();
+ UpdateDistributedConfigurationAction
updateDistributedConfigurationAction =
+ new UpdateDistributedConfigurationAction(
+ null,
+ () -> {
+ nextAction.complete(null);
+ return null;
+ }
+ );
+
+ when(cmgMgr.clusterConfigurationToUpdate())
+
.thenReturn(completedFuture(updateDistributedConfigurationAction));
+
+ // Run updater.
+ DistributedConfigurationUpdater distributedConfigurationUpdater = new
DistributedConfigurationUpdater(
+ cmgMgr,
+ presentation
+ );
+
+ distributedConfigurationUpdater.start();
+
+ // Verify that configuration wasn't updated.
+ verify(presentation, never()).update(any());
+
+ // Verify that next action is completed.
+ nextAction.join();
Review Comment:
please use `willCompleteSuccessfully()`
##########
modules/cluster-management/src/integrationTest/java/org/apache/ignite/internal/cluster/management/ItClusterManagerTest.java:
##########
@@ -351,6 +355,45 @@ void testLeaderChangeDuringJoin(TestInfo testInfo) throws
Exception {
assertThat(node.clusterManager().onJoinReady(),
willCompleteSuccessfully());
}
+ @Test
+ void
testClusterConfigurationIsRemovedFromClusterStateAfterUpdating(TestInfo
testInfo) throws Exception {
+ // Start a cluster of 3 nodes so that the CMG leader node could be
stopped later.
+ startCluster(3, testInfo);
+
+ String[] cmgNodes = clusterNodeNames();
+
+ // Start the CMG on all 3 nodes.
+ String clusterConfiguration = "security.authentication.enabled:true";
+ initCluster(cmgNodes, cmgNodes, clusterConfiguration);
+
+ // Find the CMG leader and stop it.
+ MockNode leaderNode = findLeaderNode(cluster).orElseThrow();
+
+ // Read cluster configuration from the cluster state and remove it.
+ UpdateDistributedConfigurationAction configurationAction =
leaderNode.clusterManager()
+ .clusterConfigurationToUpdate()
+ .get();
+
+ assertThat(configurationAction.configuration(),
is(clusterConfiguration));
+
configurationAction.nextAction().apply(CompletableFuture.completedFuture(null)).join();
+
+ // Stop the cluster leader.
+ stopNodes(List.of(leaderNode));
Review Comment:
This comment has not been fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
