sashapolo commented on a change in pull request #340:
URL: https://github.com/apache/ignite-3/pull/340#discussion_r714711181
##########
File path:
modules/metastorage-client/src/integrationTest/java/org/apache/ignite/internal/metastorage/client/ITMetaStorageServiceTest.java
##########
@@ -217,6 +223,8 @@ public void beforeTest() throws Exception {
LOG.info("Cluster started.");
+ executor = new ScheduledThreadPoolExecutor(20);
Review comment:
I think this line can be moved into the field initialization and the
field can be made `final`
##########
File path:
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ITLozaTest.java
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.raft;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.ClusterServiceFactory;
+import org.apache.ignite.network.MessageSerializationRegistryImpl;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.StaticNodeFinder;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.apache.ignite.raft.client.WriteCommand;
+import org.apache.ignite.raft.client.service.CommandClosure;
+import org.apache.ignite.raft.client.service.RaftGroupListener;
+import org.apache.ignite.utils.ClusterServiceTestUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.apache.ignite.raft.jraft.test.TestUtils.getLocalAddress;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link Loza} functionality.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@ExtendWith(MockitoExtension.class)
Review comment:
why do you need this extension?
##########
File path:
modules/raft/src/integrationTest/java/org/apache/ignite/raft/client/service/ITAbstractListenerSnapshotTest.java
##########
@@ -92,8 +95,19 @@
/** Clients. */
private final List<RaftGroupService> clients = new ArrayList<>();
+ /** Executor for raft group services. */
+ private ScheduledExecutorService executor;
+
+ /**
+ * Create executor for raft group services.
+ */
+ @BeforeEach
+ public void beforeTest() {
+ executor = new ScheduledThreadPoolExecutor(20);
Review comment:
same stuff: this can be replaced with field initialization
##########
File path:
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ITLozaTest.java
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.raft;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.ClusterServiceFactory;
+import org.apache.ignite.network.MessageSerializationRegistryImpl;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.StaticNodeFinder;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.apache.ignite.raft.client.WriteCommand;
+import org.apache.ignite.raft.client.service.CommandClosure;
+import org.apache.ignite.raft.client.service.RaftGroupListener;
+import org.apache.ignite.utils.ClusterServiceTestUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.apache.ignite.raft.jraft.test.TestUtils.getLocalAddress;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link Loza} functionality.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@ExtendWith(MockitoExtension.class)
+public class ITLozaTest {
+ /** Node's name. */
+ private static final String NODE_NAME = "node1";
+
+ /** Network factory. */
+ protected static final ClusterServiceFactory NETWORK_FACTORY = new
TestScaleCubeClusterServiceFactory();
+
+ /** Server port offset. */
+ protected static final int PORT = 20010;
+
+ /** */
+ private static final MessageSerializationRegistry SERIALIZATION_REGISTRY =
new MessageSerializationRegistryImpl();
+
+ /** */
+ @WorkDirectory
+ private Path dataPath;
+
+ /** Test node. */
+ private final ClusterNode node = new ClusterNode(
+ UUID.randomUUID().toString(),
+ NODE_NAME,
+ new NetworkAddress(getLocalAddress(), PORT)
+ );
+
+ /**
+ * @param groupId Raft group id.
+ */
+ private void startClient(String groupId, Loza loza) throws
ExecutionException, InterruptedException, TimeoutException {
Review comment:
I would suggest that this method returns a future. This way you can
start all of the raft groups in parallel
##########
File path:
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ITLozaTest.java
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.raft;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.ClusterServiceFactory;
+import org.apache.ignite.network.MessageSerializationRegistryImpl;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.StaticNodeFinder;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.apache.ignite.raft.client.WriteCommand;
+import org.apache.ignite.raft.client.service.CommandClosure;
+import org.apache.ignite.raft.client.service.RaftGroupListener;
+import org.apache.ignite.utils.ClusterServiceTestUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.apache.ignite.raft.jraft.test.TestUtils.getLocalAddress;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link Loza} functionality.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@ExtendWith(MockitoExtension.class)
+public class ITLozaTest {
+ /** Node's name. */
+ private static final String NODE_NAME = "node1";
+
+ /** Network factory. */
+ protected static final ClusterServiceFactory NETWORK_FACTORY = new
TestScaleCubeClusterServiceFactory();
+
+ /** Server port offset. */
+ protected static final int PORT = 20010;
+
+ /** */
+ private static final MessageSerializationRegistry SERIALIZATION_REGISTRY =
new MessageSerializationRegistryImpl();
+
+ /** */
+ @WorkDirectory
+ private Path dataPath;
+
+ /** Test node. */
+ private final ClusterNode node = new ClusterNode(
+ UUID.randomUUID().toString(),
+ NODE_NAME,
+ new NetworkAddress(getLocalAddress(), PORT)
+ );
+
+ /**
+ * @param groupId Raft group id.
+ */
+ private void startClient(String groupId, Loza loza) throws
ExecutionException, InterruptedException, TimeoutException {
+ loza.prepareRaftGroup(groupId, List.of(node), () -> new
RaftGroupListener() {
+ @Override public void onRead(Iterator<CommandClosure<ReadCommand>>
iterator) {
+ }
+
+ @Override public void
onWrite(Iterator<CommandClosure<WriteCommand>> iterator) {
+
+ }
+
+ @Override public void onSnapshotSave(Path path,
Consumer<Throwable> doneClo) {
+
+ }
+
+ @Override public boolean onSnapshotLoad(Path path) {
+ return false;
+ }
+
+ @Override public void onShutdown() {
+
+ }
+ }).get(10, TimeUnit.SECONDS);
+ }
+
+ /**
+ * @param name Node name.
+ * @param port Local port.
+ * @param servers Server nodes of the cluster.
+ * @return The client cluster view.
+ */
+ protected ClusterService clusterService(String name, int port,
List<NetworkAddress> servers) {
Review comment:
```suggestion
private static ClusterService clusterService(String name, int port,
List<NetworkAddress> servers) {
```
##########
File path:
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ITLozaTest.java
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.raft;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.ClusterServiceFactory;
+import org.apache.ignite.network.MessageSerializationRegistryImpl;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.StaticNodeFinder;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.apache.ignite.raft.client.WriteCommand;
+import org.apache.ignite.raft.client.service.CommandClosure;
+import org.apache.ignite.raft.client.service.RaftGroupListener;
+import org.apache.ignite.utils.ClusterServiceTestUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.apache.ignite.raft.jraft.test.TestUtils.getLocalAddress;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link Loza} functionality.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@ExtendWith(MockitoExtension.class)
+public class ITLozaTest {
+ /** Node's name. */
+ private static final String NODE_NAME = "node1";
+
+ /** Network factory. */
+ protected static final ClusterServiceFactory NETWORK_FACTORY = new
TestScaleCubeClusterServiceFactory();
+
+ /** Server port offset. */
+ protected static final int PORT = 20010;
+
+ /** */
+ private static final MessageSerializationRegistry SERIALIZATION_REGISTRY =
new MessageSerializationRegistryImpl();
+
+ /** */
+ @WorkDirectory
+ private Path dataPath;
+
+ /** Test node. */
+ private final ClusterNode node = new ClusterNode(
+ UUID.randomUUID().toString(),
+ NODE_NAME,
+ new NetworkAddress(getLocalAddress(), PORT)
+ );
+
+ /**
+ * @param groupId Raft group id.
+ */
+ private void startClient(String groupId, Loza loza) throws
ExecutionException, InterruptedException, TimeoutException {
+ loza.prepareRaftGroup(groupId, List.of(node), () -> new
RaftGroupListener() {
+ @Override public void onRead(Iterator<CommandClosure<ReadCommand>>
iterator) {
+ }
+
+ @Override public void
onWrite(Iterator<CommandClosure<WriteCommand>> iterator) {
+
+ }
+
+ @Override public void onSnapshotSave(Path path,
Consumer<Throwable> doneClo) {
+
+ }
+
+ @Override public boolean onSnapshotLoad(Path path) {
+ return false;
+ }
+
+ @Override public void onShutdown() {
+
+ }
+ }).get(10, TimeUnit.SECONDS);
+ }
+
+ /**
+ * @param name Node name.
+ * @param port Local port.
+ * @param servers Server nodes of the cluster.
+ * @return The client cluster view.
+ */
+ protected ClusterService clusterService(String name, int port,
List<NetworkAddress> servers) {
+ var network = ClusterServiceTestUtils.clusterService(
+ name,
+ port,
+ new StaticNodeFinder(servers),
+ SERIALIZATION_REGISTRY,
+ NETWORK_FACTORY
+ );
+
+ network.start();
+
+ return network;
+ }
+
+ /**
+ * Tests that RaftGroupServiceImpl uses shared executor for retrying
RaftGroupServiceImpl#sendWithRetry()
+ */
+ @Test
+ public void testRaftServiceUsingSharedExecutor() throws Exception {
+ var addr1 = new NetworkAddress(getLocalAddress(), PORT);
Review comment:
you can use `node.address()` instead
##########
File path:
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ITLozaTest.java
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.raft;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.ClusterServiceFactory;
+import org.apache.ignite.network.MessageSerializationRegistryImpl;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.StaticNodeFinder;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.apache.ignite.raft.client.WriteCommand;
+import org.apache.ignite.raft.client.service.CommandClosure;
+import org.apache.ignite.raft.client.service.RaftGroupListener;
+import org.apache.ignite.utils.ClusterServiceTestUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.apache.ignite.raft.jraft.test.TestUtils.getLocalAddress;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link Loza} functionality.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@ExtendWith(MockitoExtension.class)
+public class ITLozaTest {
+ /** Node's name. */
+ private static final String NODE_NAME = "node1";
+
+ /** Network factory. */
+ protected static final ClusterServiceFactory NETWORK_FACTORY = new
TestScaleCubeClusterServiceFactory();
Review comment:
```suggestion
private static final ClusterServiceFactory NETWORK_FACTORY = new
TestScaleCubeClusterServiceFactory();
```
##########
File path:
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ITLozaTest.java
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.raft;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.ClusterServiceFactory;
+import org.apache.ignite.network.MessageSerializationRegistryImpl;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.StaticNodeFinder;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.apache.ignite.raft.client.WriteCommand;
+import org.apache.ignite.raft.client.service.CommandClosure;
+import org.apache.ignite.raft.client.service.RaftGroupListener;
+import org.apache.ignite.utils.ClusterServiceTestUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.apache.ignite.raft.jraft.test.TestUtils.getLocalAddress;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link Loza} functionality.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@ExtendWith(MockitoExtension.class)
+public class ITLozaTest {
+ /** Node's name. */
+ private static final String NODE_NAME = "node1";
+
+ /** Network factory. */
+ protected static final ClusterServiceFactory NETWORK_FACTORY = new
TestScaleCubeClusterServiceFactory();
+
+ /** Server port offset. */
+ protected static final int PORT = 20010;
+
+ /** */
+ private static final MessageSerializationRegistry SERIALIZATION_REGISTRY =
new MessageSerializationRegistryImpl();
+
+ /** */
+ @WorkDirectory
+ private Path dataPath;
+
+ /** Test node. */
+ private final ClusterNode node = new ClusterNode(
+ UUID.randomUUID().toString(),
+ NODE_NAME,
+ new NetworkAddress(getLocalAddress(), PORT)
+ );
+
+ /**
+ * @param groupId Raft group id.
Review comment:
This is not a good javadoc
##########
File path:
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ITLozaTest.java
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.raft;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.ClusterServiceFactory;
+import org.apache.ignite.network.MessageSerializationRegistryImpl;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.StaticNodeFinder;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.apache.ignite.raft.client.WriteCommand;
+import org.apache.ignite.raft.client.service.CommandClosure;
+import org.apache.ignite.raft.client.service.RaftGroupListener;
+import org.apache.ignite.utils.ClusterServiceTestUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.apache.ignite.raft.jraft.test.TestUtils.getLocalAddress;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link Loza} functionality.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@ExtendWith(MockitoExtension.class)
+public class ITLozaTest {
+ /** Node's name. */
+ private static final String NODE_NAME = "node1";
+
+ /** Network factory. */
+ protected static final ClusterServiceFactory NETWORK_FACTORY = new
TestScaleCubeClusterServiceFactory();
+
+ /** Server port offset. */
+ protected static final int PORT = 20010;
Review comment:
```suggestion
private static final int PORT = 20010;
```
##########
File path:
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ITLozaTest.java
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.raft;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.ClusterServiceFactory;
+import org.apache.ignite.network.MessageSerializationRegistryImpl;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.StaticNodeFinder;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.apache.ignite.raft.client.WriteCommand;
+import org.apache.ignite.raft.client.service.CommandClosure;
+import org.apache.ignite.raft.client.service.RaftGroupListener;
+import org.apache.ignite.utils.ClusterServiceTestUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.apache.ignite.raft.jraft.test.TestUtils.getLocalAddress;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link Loza} functionality.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@ExtendWith(MockitoExtension.class)
+public class ITLozaTest {
+ /** Node's name. */
+ private static final String NODE_NAME = "node1";
+
+ /** Network factory. */
+ protected static final ClusterServiceFactory NETWORK_FACTORY = new
TestScaleCubeClusterServiceFactory();
+
+ /** Server port offset. */
+ protected static final int PORT = 20010;
+
+ /** */
+ private static final MessageSerializationRegistry SERIALIZATION_REGISTRY =
new MessageSerializationRegistryImpl();
+
+ /** */
+ @WorkDirectory
+ private Path dataPath;
+
+ /** Test node. */
+ private final ClusterNode node = new ClusterNode(
+ UUID.randomUUID().toString(),
+ NODE_NAME,
+ new NetworkAddress(getLocalAddress(), PORT)
+ );
+
+ /**
+ * @param groupId Raft group id.
+ */
+ private void startClient(String groupId, Loza loza) throws
ExecutionException, InterruptedException, TimeoutException {
+ loza.prepareRaftGroup(groupId, List.of(node), () -> new
RaftGroupListener() {
Review comment:
this stuff can be written shorter:
```
loza.prepareRaftGroup(groupId, List.of(node), () ->
mock(RaftGroupListener.class))
```
##########
File path:
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ITLozaTest.java
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.raft;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.ClusterServiceFactory;
+import org.apache.ignite.network.MessageSerializationRegistryImpl;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.StaticNodeFinder;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.apache.ignite.raft.client.WriteCommand;
+import org.apache.ignite.raft.client.service.CommandClosure;
+import org.apache.ignite.raft.client.service.RaftGroupListener;
+import org.apache.ignite.utils.ClusterServiceTestUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.apache.ignite.raft.jraft.test.TestUtils.getLocalAddress;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link Loza} functionality.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@ExtendWith(MockitoExtension.class)
+public class ITLozaTest {
+ /** Node's name. */
+ private static final String NODE_NAME = "node1";
+
+ /** Network factory. */
+ protected static final ClusterServiceFactory NETWORK_FACTORY = new
TestScaleCubeClusterServiceFactory();
+
+ /** Server port offset. */
+ protected static final int PORT = 20010;
+
+ /** */
+ private static final MessageSerializationRegistry SERIALIZATION_REGISTRY =
new MessageSerializationRegistryImpl();
+
+ /** */
+ @WorkDirectory
+ private Path dataPath;
+
+ /** Test node. */
+ private final ClusterNode node = new ClusterNode(
+ UUID.randomUUID().toString(),
+ NODE_NAME,
+ new NetworkAddress(getLocalAddress(), PORT)
+ );
+
+ /**
+ * @param groupId Raft group id.
+ */
+ private void startClient(String groupId, Loza loza) throws
ExecutionException, InterruptedException, TimeoutException {
Review comment:
This line is too long
##########
File path:
modules/raft/src/integrationTest/java/org/apache/ignite/internal/raft/ITLozaTest.java
##########
@@ -0,0 +1,190 @@
+/*
+ * 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.raft;
+
+import java.io.IOException;
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import org.apache.ignite.internal.testframework.WorkDirectory;
+import org.apache.ignite.internal.testframework.WorkDirectoryExtension;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+import org.apache.ignite.network.ClusterServiceFactory;
+import org.apache.ignite.network.MessageSerializationRegistryImpl;
+import org.apache.ignite.network.MessagingService;
+import org.apache.ignite.network.NetworkAddress;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.StaticNodeFinder;
+import org.apache.ignite.network.scalecube.TestScaleCubeClusterServiceFactory;
+import org.apache.ignite.network.serialization.MessageSerializationRegistry;
+import org.apache.ignite.raft.client.ReadCommand;
+import org.apache.ignite.raft.client.WriteCommand;
+import org.apache.ignite.raft.client.service.CommandClosure;
+import org.apache.ignite.raft.client.service.RaftGroupListener;
+import org.apache.ignite.utils.ClusterServiceTestUtils;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+import static org.apache.ignite.raft.jraft.test.TestUtils.getLocalAddress;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+/**
+ * Tests for {@link Loza} functionality.
+ */
+@ExtendWith(WorkDirectoryExtension.class)
+@ExtendWith(MockitoExtension.class)
+public class ITLozaTest {
+ /** Node's name. */
+ private static final String NODE_NAME = "node1";
+
+ /** Network factory. */
+ protected static final ClusterServiceFactory NETWORK_FACTORY = new
TestScaleCubeClusterServiceFactory();
+
+ /** Server port offset. */
+ protected static final int PORT = 20010;
+
+ /** */
+ private static final MessageSerializationRegistry SERIALIZATION_REGISTRY =
new MessageSerializationRegistryImpl();
+
+ /** */
+ @WorkDirectory
+ private Path dataPath;
+
+ /** Test node. */
+ private final ClusterNode node = new ClusterNode(
+ UUID.randomUUID().toString(),
+ NODE_NAME,
+ new NetworkAddress(getLocalAddress(), PORT)
+ );
+
+ /**
+ * @param groupId Raft group id.
+ */
+ private void startClient(String groupId, Loza loza) throws
ExecutionException, InterruptedException, TimeoutException {
+ loza.prepareRaftGroup(groupId, List.of(node), () -> new
RaftGroupListener() {
+ @Override public void onRead(Iterator<CommandClosure<ReadCommand>>
iterator) {
+ }
+
+ @Override public void
onWrite(Iterator<CommandClosure<WriteCommand>> iterator) {
+
+ }
+
+ @Override public void onSnapshotSave(Path path,
Consumer<Throwable> doneClo) {
+
+ }
+
+ @Override public boolean onSnapshotLoad(Path path) {
+ return false;
+ }
+
+ @Override public void onShutdown() {
+
+ }
+ }).get(10, TimeUnit.SECONDS);
+ }
+
+ /**
+ * @param name Node name.
+ * @param port Local port.
+ * @param servers Server nodes of the cluster.
+ * @return The client cluster view.
+ */
+ protected ClusterService clusterService(String name, int port,
List<NetworkAddress> servers) {
+ var network = ClusterServiceTestUtils.clusterService(
+ name,
+ port,
+ new StaticNodeFinder(servers),
+ SERIALIZATION_REGISTRY,
+ NETWORK_FACTORY
+ );
+
+ network.start();
+
+ return network;
+ }
+
+ /**
+ * Tests that RaftGroupServiceImpl uses shared executor for retrying
RaftGroupServiceImpl#sendWithRetry()
+ */
+ @Test
+ public void testRaftServiceUsingSharedExecutor() throws Exception {
+ var addr1 = new NetworkAddress(getLocalAddress(), PORT);
+
+ ClusterService service = spy(clusterService(node.name(), PORT,
List.of(addr1)));
+
+ MessagingService messagingServiceMock = mock(MessagingService.class);
+
+ when(service.messagingService()).thenReturn(messagingServiceMock);
+
+ CompletableFuture<NetworkMessage> fut = new CompletableFuture<>();
+
+ AtomicInteger executorsInvocations = new AtomicInteger(0);
+
+ when(messagingServiceMock.invoke((NetworkAddress)any(), any(),
anyLong())).thenAnswer(mock -> {
+ String threadName = Thread.currentThread().getName();
+
+ assertTrue(threadName.contains(Loza.CLIENT_POOL_NAME) ||
threadName.contains("main"));
+
+ if (threadName.contains(Loza.CLIENT_POOL_NAME))
+ executorsInvocations.incrementAndGet();
+
+ return fut;
+ });
+
+ // Need to complete exceptionally to enforce sendWithRetry to be
scheduled with shared executor.
+ // See RaftGroupServiceImpl#recoverable
+ fut.completeExceptionally(new Exception("Test exception", new
IOException()));
+
+ Loza loza = new Loza(service, dataPath);
+
+ loza.start();
+
+ for (int i = 0; i < 5; i++) {
+ try {
+ startClient(Integer.toString(i), loza);
+ }
+ catch (ExecutionException | InterruptedException |
TimeoutException e) {
+ if (!(e.getCause() instanceof TimeoutException))
+ fail(e);
+
+ assertTrue(executorsInvocations.get() > 1);
+ }
+
+ executorsInvocations.set(0);
+ }
+
+ loza.stop();
Review comment:
This stuff should be declared in either `finally` block or in
`@AfterEach`
##########
File path:
modules/raft/src/integrationTest/java/org/apache/ignite/raft/server/ITJRaftCounterServerTest.java
##########
@@ -126,10 +128,15 @@
@WorkDirectory
private Path dataPath;
+ /** Executor for raft group services. */
+ private ScheduledExecutorService executor;
+
/** */
@BeforeEach
void before(TestInfo testInfo) {
LOG.info(">>>>>>>>>>>>>>> Start test method: {}",
testInfo.getTestMethod().orElseThrow().getName());
+
+ executor = new ScheduledThreadPoolExecutor(20);
Review comment:
and here
##########
File path: modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java
##########
@@ -61,6 +78,8 @@ public Loza(ClusterService clusterNetSvc, Path dataPath) {
this.clusterNetSvc = clusterNetSvc;
this.raftServer = new JRaftServerImpl(clusterNetSvc, dataPath);
+
+ this.executor = new ScheduledThreadPoolExecutor(CLIENT_POOL_SIZE, new
NamedThreadFactory(NamedThreadFactory.threadPrefix(clusterNetSvc.localConfiguration().getName(),
CLIENT_POOL_NAME)));
Review comment:
too long line
##########
File path:
modules/table/src/integrationTest/java/org/apache/ignite/distributed/ITDistributedTableTest.java
##########
@@ -200,6 +205,8 @@ public void partitionListener() throws Exception {
assertEquals(testRow.longValue(1), new Row(SCHEMA,
getFut.get().getValue()).longValue(1));
partSrv.stop();
+
+ IgniteUtils.shutdownAndAwaitTermination(executor, 10,
TimeUnit.SECONDS);
Review comment:
should be in either `finally` or `AfterEach` block
##########
File path: modules/raft/src/main/java/org/apache/ignite/internal/raft/Loza.java
##########
@@ -52,6 +66,9 @@
/** Raft server. */
private final RaftServer raftServer;
+ /** Executor for raft group services. */
+ private ScheduledExecutorService executor;
Review comment:
```suggestion
private final ScheduledExecutorService executor;
```
--
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]
