This is an automated email from the ASF dual-hosted git repository.
apkhmv pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 3acc9bca4b IGNITE-18551 Create general task executor for all
registries (#2060)
3acc9bca4b is described below
commit 3acc9bca4bb91aae7b3485e85d03efc57ae41673
Author: Vadim Pakhnushev <[email protected]>
AuthorDate: Thu May 11 17:11:37 2023 +0300
IGNITE-18551 Create general task executor for all registries (#2060)
---
.../repl/executor/ItIgnitePicocliCommandsTest.java | 3 +-
.../internal/cli/commands/node/NodeUrlMixin.java | 1 -
.../core/converters/NodeNameOrUrlConverter.java | 4 +-
...eNameRegistry.java => PeriodicSessionTask.java} | 22 +++----
.../cli/core/repl/PeriodicSessionTaskExecutor.java | 66 ++++++++++++++++++++
.../cluster/ClusterUrlDynamicCompleterFactory.java | 8 +--
.../cli/core/repl/registry/NodeNameRegistry.java | 5 +-
.../repl/registry/impl/JdbcUrlRegistryImpl.java | 44 ++++----------
.../repl/registry/impl/NodeNameRegistryImpl.java | 70 +++++++---------------
9 files changed, 114 insertions(+), 109 deletions(-)
diff --git
a/modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/core/repl/executor/ItIgnitePicocliCommandsTest.java
b/modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/core/repl/executor/ItIgnitePicocliCommandsTest.java
index 2586a36d0e..1c6702c0b0 100644
---
a/modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/core/repl/executor/ItIgnitePicocliCommandsTest.java
+++
b/modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/core/repl/executor/ItIgnitePicocliCommandsTest.java
@@ -29,7 +29,6 @@ import static org.mockito.Mockito.when;
import jakarta.inject.Inject;
import java.io.File;
-import java.net.URL;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -396,7 +395,7 @@ public class ItIgnitePicocliCommandsTest extends
CliCommandTestInitializedIntegr
await().until(() -> nodeNameRegistry.urls(), not(empty()));
// Then
- String[] expectedUrls =
nodeNameRegistry.urls().stream().map(URL::toString).toArray(String[]::new);
+ String[] expectedUrls = nodeNameRegistry.urls().toArray(String[]::new);
assertThat("For given parsed words: " + parsedLine.words(),
complete(parsedLine),
containsInAnyOrder(expectedUrls));
diff --git
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/commands/node/NodeUrlMixin.java
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/commands/node/NodeUrlMixin.java
index a85707217c..8b408dfdc4 100644
---
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/commands/node/NodeUrlMixin.java
+++
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/commands/node/NodeUrlMixin.java
@@ -71,7 +71,6 @@ public class NodeUrlMixin {
}
if (options.nodeName != null) {
return nodeNameRegistry.nodeUrlByName(options.nodeName)
- .map(URL::toString)
.orElseThrow(() -> new IgniteCliException("Node " +
options.nodeName
+ " not found. Provide a valid name or use a
URL"));
}
diff --git
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/converters/NodeNameOrUrlConverter.java
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/converters/NodeNameOrUrlConverter.java
index 0da16ff179..558a4fbb8b 100644
---
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/converters/NodeNameOrUrlConverter.java
+++
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/converters/NodeNameOrUrlConverter.java
@@ -50,11 +50,11 @@ public class NodeNameOrUrlConverter implements
CommandLine.ITypeConverter<NodeNa
if (isUrl) {
return new NodeNameOrUrl(stringToUrl(input));
} else {
- return new NodeNameOrUrl(findNodeUrlByNodeName(input));
+ return new
NodeNameOrUrl(stringToUrl(findNodeUrlByNodeName(input)));
}
}
- private URL findNodeUrlByNodeName(String name) {
+ private String findNodeUrlByNodeName(String name) {
return nodeNameRegistry.nodeUrlByName(name)
.orElseThrow(() -> new TypeConversionException("Node " + name
+ " not found. Provide valid name or use URL"));
}
diff --git
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/NodeNameRegistry.java
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/PeriodicSessionTask.java
similarity index 65%
copy from
modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/NodeNameRegistry.java
copy to
modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/PeriodicSessionTask.java
index 4c267b1311..2b05693559 100644
---
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/NodeNameRegistry.java
+++
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/PeriodicSessionTask.java
@@ -15,22 +15,14 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.cli.core.repl.registry;
+package org.apache.ignite.internal.cli.core.repl;
-import java.net.URL;
-import java.util.Optional;
-import java.util.Set;
+/** Task which will be executed periodically while {@link Session} is
connected. */
+public interface PeriodicSessionTask {
-/** Node name registry. */
-public interface NodeNameRegistry {
-
- /** Returns a node url by a provided node name. */
- Optional<URL> nodeUrlByName(String nodeName);
-
- /** Returns set of node names. */
- Set<String> names();
-
- /** Returns set of node urls. */
- Set<URL> urls();
+ /** This method is called periodically. */
+ void update(SessionInfo sessionInfo);
+ /** This method is called when the session is disconnected. */
+ void onDisconnect();
}
diff --git
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/PeriodicSessionTaskExecutor.java
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/PeriodicSessionTaskExecutor.java
new file mode 100644
index 0000000000..364c49abe6
--- /dev/null
+++
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/PeriodicSessionTaskExecutor.java
@@ -0,0 +1,66 @@
+/*
+ * 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.cli.core.repl;
+
+import static
org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
+
+import jakarta.inject.Singleton;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.internal.cli.logger.CliLoggers;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.thread.NamedThreadFactory;
+import org.jetbrains.annotations.Nullable;
+
+/** Executes tasks periodically while the session is connected. */
+@Singleton
+public class PeriodicSessionTaskExecutor implements AsyncSessionEventListener {
+ private static final IgniteLogger LOG =
CliLoggers.forClass(PeriodicSessionTaskExecutor.class);
+
+ @Nullable
+ private ScheduledExecutorService executor;
+
+ private final List<? extends PeriodicSessionTask> tasks;
+
+ public PeriodicSessionTaskExecutor(List<? extends PeriodicSessionTask>
tasks) {
+ this.tasks = tasks;
+ }
+
+ @Override
+ public synchronized void onConnect(SessionInfo sessionInfo) {
+ if (executor == null) {
+ executor = Executors.newSingleThreadScheduledExecutor(new
NamedThreadFactory("SessionTaskExecutor", LOG));
+ executor.scheduleWithFixedDelay(() -> runTasks(sessionInfo), 0, 5,
TimeUnit.SECONDS);
+ }
+ }
+
+ @Override
+ public synchronized void onDisconnect() {
+ if (executor != null) {
+ shutdownAndAwaitTermination(executor, 3, TimeUnit.SECONDS);
+ tasks.forEach(PeriodicSessionTask::onDisconnect);
+ executor = null;
+ }
+ }
+
+ private void runTasks(SessionInfo sessionInfo) {
+ tasks.forEach(periodicSessionTask ->
periodicSessionTask.update(sessionInfo));
+ }
+}
diff --git
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/completer/cluster/ClusterUrlDynamicCompleterFactory.java
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/completer/cluster/ClusterUrlDynamicCompleterFactory.java
index 39cf583080..59405ac824 100644
---
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/completer/cluster/ClusterUrlDynamicCompleterFactory.java
+++
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/completer/cluster/ClusterUrlDynamicCompleterFactory.java
@@ -18,9 +18,8 @@
package org.apache.ignite.internal.cli.core.repl.completer.cluster;
import jakarta.inject.Singleton;
-import java.net.URL;
+import java.util.HashSet;
import java.util.Set;
-import java.util.stream.Collectors;
import org.apache.ignite.internal.cli.core.repl.completer.DynamicCompleter;
import
org.apache.ignite.internal.cli.core.repl.completer.DynamicCompleterFactory;
import
org.apache.ignite.internal.cli.core.repl.completer.StringDynamicCompleter;
@@ -38,10 +37,7 @@ public class ClusterUrlDynamicCompleterFactory implements
DynamicCompleterFactor
@Override
public DynamicCompleter getDynamicCompleter(String[] words) {
- Set<String> urls = nodeNameRegistry.urls()
- .stream()
- .map(URL::toString)
- .collect(Collectors.toSet());
+ Set<String> urls = new HashSet<>(nodeNameRegistry.urls());
return new StringDynamicCompleter(urls);
}
}
diff --git
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/NodeNameRegistry.java
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/NodeNameRegistry.java
index 4c267b1311..7622c5c598 100644
---
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/NodeNameRegistry.java
+++
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/NodeNameRegistry.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.cli.core.repl.registry;
-import java.net.URL;
import java.util.Optional;
import java.util.Set;
@@ -25,12 +24,12 @@ import java.util.Set;
public interface NodeNameRegistry {
/** Returns a node url by a provided node name. */
- Optional<URL> nodeUrlByName(String nodeName);
+ Optional<String> nodeUrlByName(String nodeName);
/** Returns set of node names. */
Set<String> names();
/** Returns set of node urls. */
- Set<URL> urls();
+ Set<String> urls();
}
diff --git
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/impl/JdbcUrlRegistryImpl.java
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/impl/JdbcUrlRegistryImpl.java
index aa3711fab3..8dc499df19 100644
---
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/impl/JdbcUrlRegistryImpl.java
+++
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/impl/JdbcUrlRegistryImpl.java
@@ -17,33 +17,27 @@
package org.apache.ignite.internal.cli.core.repl.registry.impl;
-import static
org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
-
import jakarta.inject.Singleton;
-import java.net.URL;
import java.util.Objects;
import java.util.Set;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import org.apache.ignite.internal.cli.core.JdbcUrlFactory;
-import org.apache.ignite.internal.cli.core.repl.AsyncSessionEventListener;
+import org.apache.ignite.internal.cli.core.repl.PeriodicSessionTask;
import org.apache.ignite.internal.cli.core.repl.SessionInfo;
import org.apache.ignite.internal.cli.core.repl.registry.JdbcUrlRegistry;
import org.apache.ignite.internal.cli.core.repl.registry.NodeNameRegistry;
import org.apache.ignite.internal.cli.core.rest.ApiClientFactory;
import org.apache.ignite.internal.cli.logger.CliLoggers;
import org.apache.ignite.internal.logger.IgniteLogger;
-import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.rest.client.api.NodeConfigurationApi;
import org.apache.ignite.rest.client.invoker.ApiException;
+import org.jetbrains.annotations.Nullable;
/** Implementation of {@link JdbcUrlRegistry}. */
@Singleton
-public class JdbcUrlRegistryImpl implements JdbcUrlRegistry,
AsyncSessionEventListener {
+public class JdbcUrlRegistryImpl implements JdbcUrlRegistry,
PeriodicSessionTask {
- private static final IgniteLogger log =
CliLoggers.forClass(JdbcUrlRegistryImpl.class);
+ private static final IgniteLogger LOG =
CliLoggers.forClass(JdbcUrlRegistryImpl.class);
private final NodeNameRegistry nodeNameRegistry;
@@ -53,8 +47,6 @@ public class JdbcUrlRegistryImpl implements JdbcUrlRegistry,
AsyncSessionEventLi
private volatile Set<String> jdbcUrls = Set.of();
- private ScheduledExecutorService executor;
-
/** Constructor. */
public JdbcUrlRegistryImpl(NodeNameRegistry nodeNameRegistry,
ApiClientFactory clientFactory, JdbcUrlFactory jdbcUrlFactory) {
this.nodeNameRegistry = nodeNameRegistry;
@@ -62,26 +54,32 @@ public class JdbcUrlRegistryImpl implements
JdbcUrlRegistry, AsyncSessionEventLi
this.jdbcUrlFactory = jdbcUrlFactory;
}
- private void fetchJdbcUrls() {
+ @Override
+ public void update(SessionInfo sessionInfo) {
jdbcUrls = nodeNameRegistry.urls()
.stream()
- .map(URL::toString)
.map(this::fetchJdbcUrl)
.filter(Objects::nonNull)
.collect(Collectors.toSet());
}
+ @Override
+ public void onDisconnect() {
+ jdbcUrls = Set.of();
+ }
+
/** {@inheritDoc} */
@Override
public Set<String> jdbcUrls() {
return Set.copyOf(jdbcUrls);
}
+ @Nullable
private String fetchJdbcUrl(String nodeUrl) {
try {
return
jdbcUrlFactory.constructJdbcUrl(fetchNodeConfiguration(nodeUrl), nodeUrl);
} catch (ApiException e) {
- log.warn("Couldn't fetch jdbc url of " + nodeUrl + " node: ", e);
+ LOG.warn("Couldn't fetch jdbc url of " + nodeUrl + " node: ", e);
return null;
}
}
@@ -89,20 +87,4 @@ public class JdbcUrlRegistryImpl implements JdbcUrlRegistry,
AsyncSessionEventLi
private String fetchNodeConfiguration(String nodeUrl) throws ApiException {
return new
NodeConfigurationApi(clientFactory.getClient(nodeUrl)).getNodeConfiguration();
}
-
- @Override
- public synchronized void onConnect(SessionInfo sessionInfo) {
- if (executor == null) {
- executor = Executors.newSingleThreadScheduledExecutor(new
NamedThreadFactory("JdbcUrlRegistry", log));
- executor.scheduleWithFixedDelay(this::fetchJdbcUrls, 0, 5,
TimeUnit.SECONDS);
- }
- }
-
- @Override
- public synchronized void onDisconnect() {
- if (executor != null) {
- shutdownAndAwaitTermination(executor, 3, TimeUnit.SECONDS);
- executor = null;
- }
- }
}
diff --git
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/impl/NodeNameRegistryImpl.java
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/impl/NodeNameRegistryImpl.java
index a38b3a7193..b2657a28a1 100644
---
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/impl/NodeNameRegistryImpl.java
+++
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/core/repl/registry/impl/NodeNameRegistryImpl.java
@@ -17,41 +17,33 @@
package org.apache.ignite.internal.cli.core.repl.registry.impl;
-import static
org.apache.ignite.internal.util.IgniteUtils.shutdownAndAwaitTermination;
-
import jakarta.inject.Singleton;
import java.net.URL;
import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
-import java.util.concurrent.Executors;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import
org.apache.ignite.internal.cli.call.cluster.topology.PhysicalTopologyCall;
import org.apache.ignite.internal.cli.core.call.UrlCallInput;
-import org.apache.ignite.internal.cli.core.repl.AsyncSessionEventListener;
+import org.apache.ignite.internal.cli.core.repl.PeriodicSessionTask;
import org.apache.ignite.internal.cli.core.repl.SessionInfo;
import org.apache.ignite.internal.cli.core.repl.registry.NodeNameRegistry;
import org.apache.ignite.internal.cli.logger.CliLoggers;
import org.apache.ignite.internal.logger.IgniteLogger;
-import org.apache.ignite.internal.thread.NamedThreadFactory;
import org.apache.ignite.rest.client.model.ClusterNode;
import org.apache.ignite.rest.client.model.NodeMetadata;
import org.jetbrains.annotations.Nullable;
/** Implementation of {@link NodeNameRegistry}. */
@Singleton
-public class NodeNameRegistryImpl implements NodeNameRegistry,
AsyncSessionEventListener {
+public class NodeNameRegistryImpl implements NodeNameRegistry,
PeriodicSessionTask {
- private static final IgniteLogger log =
CliLoggers.forClass(NodeNameRegistryImpl.class);
+ private static final IgniteLogger LOG =
CliLoggers.forClass(NodeNameRegistryImpl.class);
private final PhysicalTopologyCall physicalTopologyCall;
- private volatile Map<String, URL> nodeNameToNodeUrl = Map.of();
-
- private ScheduledExecutorService executor;
+ private volatile Map<String, String> nodeNameToNodeUrl = Map.of();
public NodeNameRegistryImpl(PhysicalTopologyCall physicalTopologyCall) {
this.physicalTopologyCall = physicalTopologyCall;
@@ -59,7 +51,7 @@ public class NodeNameRegistryImpl implements
NodeNameRegistry, AsyncSessionEvent
/** {@inheritDoc} */
@Override
- public Optional<URL> nodeUrlByName(String nodeName) {
+ public Optional<String> nodeUrlByName(String nodeName) {
return Optional.ofNullable(nodeNameToNodeUrl.get(nodeName));
}
@@ -71,67 +63,47 @@ public class NodeNameRegistryImpl implements
NodeNameRegistry, AsyncSessionEvent
/** {@inheritDoc} */
@Override
- public Set<URL> urls() {
+ public Set<String> urls() {
return new HashSet<>(nodeNameToNodeUrl.values());
}
- private void updateNodeNames(String nodeUrl) {
- nodeNameToNodeUrl = physicalTopologyCall.execute(new
UrlCallInput(nodeUrl))
+ @Override
+ public void update(SessionInfo sessionInfo) {
+ nodeNameToNodeUrl = physicalTopologyCall.execute(new
UrlCallInput(sessionInfo.nodeUrl()))
.body()
.stream()
- .map(this::toNodeNameAndUrlPair)
+ .map(NodeNameRegistryImpl::toNodeNameAndUrlPair)
.filter(it -> it.url != null)
.collect(Collectors.toUnmodifiableMap(it -> it.name, it ->
it.url));
}
- private NodeNameAndUrlPair toNodeNameAndUrlPair(ClusterNode node) {
+ @Override
+ public void onDisconnect() {
+ nodeNameToNodeUrl = Map.of();
+ }
+
+ private static NodeNameAndUrlPair toNodeNameAndUrlPair(ClusterNode node) {
return new NodeNameAndUrlPair(node.getName(),
urlFromClusterNode(node.getMetadata()));
}
@Nullable
- private URL urlFromClusterNode(NodeMetadata metadata) {
+ private static String urlFromClusterNode(NodeMetadata metadata) {
if (metadata == null) {
return null;
}
try {
- return new URL("http://"; + metadata.getRestHost() + ":" +
metadata.getHttpPort());
+ return new URL("http://"; + metadata.getRestHost() + ":" +
metadata.getHttpPort()).toString();
} catch (Exception e) {
- log.warn("Couldn't create URL: {}", e);
+ LOG.warn("Couldn't create URL: {}", e);
return null;
}
}
- /**
- * Start pulling updates from a node.
- *
- * @param sessionInfo sessionInfo.
- */
-
- @Override
- public synchronized void onConnect(SessionInfo sessionInfo) {
- if (executor == null) {
- executor = Executors.newSingleThreadScheduledExecutor(new
NamedThreadFactory("NodeNameRegistry", log));
- executor.scheduleWithFixedDelay(() ->
- updateNodeNames(sessionInfo.nodeUrl()), 0, 5,
TimeUnit.SECONDS);
- }
- }
-
- /**
- * Stops pulling updates.
- */
- @Override
- public synchronized void onDisconnect() {
- if (executor != null) {
- shutdownAndAwaitTermination(executor, 3, TimeUnit.SECONDS);
- executor = null;
- }
- }
-
private static class NodeNameAndUrlPair {
private final String name;
- private final URL url;
+ private final String url;
- private NodeNameAndUrlPair(String name, @Nullable URL url) {
+ private NodeNameAndUrlPair(String name, @Nullable String url) {
this.name = name;
this.url = url;
}
