You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by mp...@apache.org on 2023/11/09 18:28:27 UTC
(ignite) branch master updated: IGNITE-20656 Java thin: Service awareness (#11003)
This is an automated email from the ASF dual-hosted git repository.
mpetrov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 7bcf23279d9 IGNITE-20656 Java thin: Service awareness (#11003)
7bcf23279d9 is described below
commit 7bcf23279d99fb7d24b6f758ef3fe8ca58f45139
Author: Vladimir Steshin <vl...@gmail.com>
AuthorDate: Thu Nov 9 21:28:18 2023 +0300
IGNITE-20656 Java thin: Service awareness (#11003)
Co-authored-by: Pavel Tupitsyn <pt...@apache.org>
---
docs/_docs/services/services.adoc | 16 +
docs/_docs/thin-clients/java-thin-client.adoc | 9 +-
.../apache/ignite/util/GridCommandHandlerTest.java | 4 +-
.../client/thin/ClientCacheAffinityContext.java | 7 +
.../internal/client/thin/ClientOperation.java | 3 +
.../internal/client/thin/ClientServicesImpl.java | 123 +++-
.../client/thin/ProtocolBitmaskFeature.java | 5 +-
.../internal/client/thin/ReliableChannel.java | 32 +-
.../internal/client/thin/TcpIgniteClient.java | 9 +-
.../platform/client/ClientBitmaskFeature.java | 5 +-
.../platform/client/ClientConnectionContext.java | 4 -
.../platform/client/ClientMessageParser.java | 7 +
.../service/ClientServiceMappingsResponse.java | 55 ++
.../service/ClientServiceTopologyRequest.java | 63 +++
.../processors/service/IgniteServiceProcessor.java | 2 +-
.../processors/service/ServiceDeploymentTask.java | 2 +-
.../org/apache/ignite/client/ReliabilityTest.java | 2 +-
.../internal/client/thin/ServiceAwarenessTest.java | 622 +++++++++++++++++++++
.../util/future/GridFutureAdapterSelfTest.java | 2 +-
.../org/apache/ignite/client/ClientTestSuite.java | 2 +
20 files changed, 951 insertions(+), 23 deletions(-)
diff --git a/docs/_docs/services/services.adoc b/docs/_docs/services/services.adoc
index ce72603e908..7568b9d6c46 100644
--- a/docs/_docs/services/services.adoc
+++ b/docs/_docs/services/services.adoc
@@ -234,6 +234,22 @@ tab:C++[]
//== Accessing Services from Compute Tasks
// TODO the @ServiceResource annotation
+== Service Awareness [[service_awareness]]
+For link:../thin-clients/java-thin-client.adoc#java_thin_client[Java Thin Client] you can activate Service Awareness.
+To do that, enable link:../thin-clients/java-thin-client.adoc#partition_awareness[Partition Awareness].
+
+Without Service Awareness, the invocation requests are sent to a random node. If it has no service
+instance deployed, the request is redirected to a different node. This additional network hop adds overhead.
+
+With Service Awareness, the thin client knows where service instances are deployed and sends the request to the correct node.
+
+[NOTE]
+====
+The service topology is updated asynchronously starting with the first service invocation.
+Thus, some invocation redirects are still possible.
+====
+
+
== Un-deploying Services
To undeploy a service, use the `IgniteServices.cancel(serviceName)` or `IgniteServices.cancelAll()` methods.
diff --git a/docs/_docs/thin-clients/java-thin-client.adoc b/docs/_docs/thin-clients/java-thin-client.adoc
index 5f955d35615..5a744d545c4 100644
--- a/docs/_docs/thin-clients/java-thin-client.adoc
+++ b/docs/_docs/thin-clients/java-thin-client.adoc
@@ -12,7 +12,7 @@
// 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.
-= Java Thin Client
+= Java Thin Client [[java_thin_client]]
:sourceCodeFile: {javaCodeDir}/JavaThinClient.java
== Overview
@@ -75,7 +75,7 @@ include::{sourceCodeFile}[tag=connect-to-many-nodes,indent=0]
Note that the code above provides a failover mechanism in case of server node failures. Refer to the <<Handling Node Failures>> section for more information.
-== Partition Awareness
+== Partition Awareness [partition_awareness]
include::includes/partition-awareness.adoc[]
@@ -116,6 +116,11 @@ The code snippet shows how an example implementation might look like if you want
Also, you can check a link:https://github.com/apache/ignite/blob/master/examples/src/main/java/org/apache/ignite/examples/client/ClientKubernetesPutGetExample.java#L50[real example] of the interface implementation. `ThinClientKubernetesAddressFinder` is created to handle scalable Kubernetes environment.
+[NOTE]
+====
+Partition Awareness also enables link:../services/services.adoc#service_awareness[Service Awareness]
+====
+
== Using Key-Value API
The Java thin client supports most of the key-value operations available in the thick client.
diff --git a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
index 6a40bfb738e..f3869dbac24 100644
--- a/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
+++ b/modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java
@@ -2053,14 +2053,14 @@ public class GridCommandHandlerTest extends GridCommandHandlerClusterPerMethodAb
String testOutStr = testOut.toString();
// Ignite instase 1 can be logged only in arguments list.
- boolean isInstanse1Found = Arrays.stream(testOutStr.split("\n"))
+ boolean isInstance1Found = Arrays.stream(testOutStr.split("\n"))
.filter(s -> s.contains("Arguments:"))
.noneMatch(s -> s.contains(getTestIgniteInstanceName() + "1"));
assertTrue(testOutStr, testOutStr.contains("Node not found for consistent ID:"));
if (commandHandler.equals(CLI_CMD_HND))
- assertFalse(testOutStr, isInstanse1Found);
+ assertFalse(testOutStr, isInstance1Found);
}
/** */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientCacheAffinityContext.java b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientCacheAffinityContext.java
index 577a2c5f476..bcd681796da 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientCacheAffinityContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientCacheAffinityContext.java
@@ -329,6 +329,13 @@ public class ClientCacheAffinityContext {
public Iterable<UUID> nodes() {
return Collections.unmodifiableCollection(nodes);
}
+
+ /**
+ * @return Topology version.
+ */
+ public AffinityTopologyVersion version() {
+ return topVer;
+ }
}
/** Holder of a mapper factory and cacheName. */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientOperation.java b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientOperation.java
index a683154a235..c97ae47928a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientOperation.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientOperation.java
@@ -216,6 +216,9 @@ public enum ClientOperation {
/** Get service descriptors. */
SERVICE_GET_DESCRIPTOR(7002),
+ /** Get service topology. */
+ SERVICE_GET_TOPOLOGY(7003),
+
/** Get or create an AtomicLong by name. */
ATOMIC_LONG_CREATE(9000),
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientServicesImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientServicesImpl.java
index f9fcb37d1fe..869c8db8883 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientServicesImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ClientServicesImpl.java
@@ -23,8 +23,13 @@ import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.IgniteLogger;
import org.apache.ignite.client.ClientClusterGroup;
import org.apache.ignite.client.ClientException;
import org.apache.ignite.client.ClientFeatureNotSupportedByServerException;
@@ -32,9 +37,11 @@ import org.apache.ignite.client.ClientServiceDescriptor;
import org.apache.ignite.client.ClientServices;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.binary.BinaryReaderExImpl;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
import org.apache.ignite.internal.processors.service.ServiceCallContextImpl;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.A;
+import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.platform.PlatformServiceMethod;
import org.apache.ignite.platform.PlatformType;
import org.apache.ignite.services.ServiceCallContext;
@@ -45,6 +52,9 @@ import org.jetbrains.annotations.Nullable;
* Implementation of {@link ClientServices}.
*/
class ClientServicesImpl implements ClientServices {
+ /** Max service topology update period in mills. */
+ static final int SRV_TOP_UPDATE_PERIOD = 10_000;
+
/** Channel. */
private final ReliableChannel ch;
@@ -57,13 +67,23 @@ class ClientServicesImpl implements ClientServices {
/** Cluster group. */
private final ClientClusterGroupImpl grp;
+ /** Logger. */
+ private final IgniteLogger log;
+
+ /** Services topology. {@code Null} if partition awareness is not enabled. */
+ private final Map<String, ServiceTopology> servicesTopologies;
+
/** Constructor. */
- ClientServicesImpl(ReliableChannel ch, ClientBinaryMarshaller marsh, ClientClusterGroupImpl grp) {
+ ClientServicesImpl(ReliableChannel ch, ClientBinaryMarshaller marsh, ClientClusterGroupImpl grp, IgniteLogger log) {
this.ch = ch;
this.marsh = marsh;
this.grp = grp;
utils = new ClientUtils(marsh);
+
+ this.log = log;
+
+ servicesTopologies = ch.partitionAwarenessEnabled ? new ConcurrentHashMap<>() : Collections.emptyMap();
}
/** {@inheritDoc} */
@@ -165,7 +185,91 @@ class ClientServicesImpl implements ClientServices {
ClientServices withClusterGroup(ClientClusterGroupImpl grp) {
A.notNull(grp, "grp");
- return new ClientServicesImpl(ch, marsh, grp);
+ return new ClientServicesImpl(ch, marsh, grp, log);
+ }
+
+ /**
+ * Keeps topology of certain service and its update progress meta.
+ */
+ private final class ServiceTopology {
+ /** The service name. */
+ private final String srvcName;
+
+ /** If {@code true}, topology update of current service is in progress. */
+ private final AtomicBoolean updateInProgress = new AtomicBoolean();
+
+ /** Time of the last received topology. */
+ private volatile long lastUpdateRequestTime;
+
+ /** UUID of the nodes with at least one service instance. */
+ private volatile List<UUID> nodes = Collections.emptyList();
+
+ /** Last cluster topology version when current service topology was actual. */
+ private volatile AffinityTopologyVersion lastAffTop;
+
+ /** */
+ private ServiceTopology(String name) {
+ srvcName = name;
+ }
+
+ /**
+ * @return {@code True} if update of the service topology is required. {@code False} otherwise.
+ */
+ private boolean isUpdateRequired(AffinityTopologyVersion curAffTop) {
+ return lastAffTop == null || curAffTop.topologyVersion() > lastAffTop.topologyVersion()
+ || U.nanosToMillis(System.nanoTime() - lastUpdateRequestTime) >= SRV_TOP_UPDATE_PERIOD;
+ }
+
+ /**
+ * Asynchronously updates the service topology.
+ */
+ private void updateTopologyAsync() {
+ AffinityTopologyVersion curAffTop = ch.affinityContext().lastTopology().version();
+
+ if (!updateInProgress.compareAndSet(false, true))
+ return;
+
+ ch.serviceAsync(
+ ClientOperation.SERVICE_GET_TOPOLOGY,
+ req -> utils.writeObject(req.out(), srvcName),
+ resp -> {
+ int cnt = resp.in().readInt();
+
+ List<UUID> res = new ArrayList<>(cnt);
+
+ for (int i = 0; i < cnt; ++i)
+ res.add(new UUID(resp.in().readLong(), resp.in().readLong()));
+
+ return res;
+ }).whenComplete((nodes, err) -> {
+ if (err == null) {
+ this.nodes = nodes;
+ lastAffTop = curAffTop;
+ lastUpdateRequestTime = System.nanoTime();
+
+ if (log.isDebugEnabled()) {
+ log.debug("Topology of service '" + srvcName + "' has been updated. The " +
+ "service instance nodes: " + nodes);
+ }
+ }
+ else
+ log.error("Failed to update topology of the service '" + srvcName + "'.", err);
+
+ updateInProgress.set(false);
+ });
+ }
+
+ /**
+ * Provides last known service topology and asynchronously updates it if required.
+ *
+ * @return Last known service topology.
+ */
+ public List<UUID> getAndUpdate() {
+ if (isUpdateRequired(ch.affinityContext().lastTopology().version()))
+ updateTopologyAsync();
+
+ return nodes;
+ }
}
/**
@@ -215,7 +319,8 @@ class ClientServicesImpl implements ClientServices {
return ch.service(ClientOperation.SERVICE_INVOKE,
req -> writeServiceInvokeRequest(req, nodeIds, method, args),
- res -> utils.readObject(res.in(), false, method.getReturnType())
+ res -> utils.readObject(res.in(), false, method.getReturnType()),
+ serviceTopology()
);
}
catch (ClientError e) {
@@ -223,6 +328,18 @@ class ClientServicesImpl implements ClientServices {
}
}
+ /**
+ * @return Actual known service topology or empty list if: service topology is not enabled, not supported or
+ * not received yet.
+ */
+ private List<UUID> serviceTopology() {
+ if (!ch.partitionAwarenessEnabled
+ || !ch.applyOnDefaultChannel(c -> c.protocolCtx().isFeatureSupported(ProtocolBitmaskFeature.SERVICE_TOPOLOGY), null))
+ return Collections.emptyList();
+
+ return servicesTopologies.computeIfAbsent(name, ServiceTopology::new).getAndUpdate();
+ }
+
/**
* @param ch Payload output channel.
* @param nodeIds Node IDs.
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ProtocolBitmaskFeature.java b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ProtocolBitmaskFeature.java
index 751f66dbc27..7676fcc323c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ProtocolBitmaskFeature.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ProtocolBitmaskFeature.java
@@ -74,7 +74,10 @@ public enum ProtocolBitmaskFeature {
INDEX_QUERY(14),
/** IndexQuery limit. */
- INDEX_QUERY_LIMIT(15);
+ INDEX_QUERY_LIMIT(15),
+
+ /** Service topology. */
+ SERVICE_TOPOLOGY(16);
/** */
private static final EnumSet<ProtocolBitmaskFeature> ALL_FEATURES_AS_ENUM_SET =
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
index 2bcf6b116c2..f88a90e654e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/ReliableChannel.java
@@ -20,6 +20,7 @@ package org.apache.ignite.internal.client.thin;
import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.Collection;
+import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
@@ -74,7 +75,7 @@ final class ReliableChannel implements AutoCloseable {
private volatile int curChIdx = -1;
/** Partition awareness enabled. */
- private final boolean partitionAwarenessEnabled;
+ final boolean partitionAwarenessEnabled;
/** Cache partition awareness context. */
private final ClientCacheAffinityContext affinityCtx;
@@ -182,7 +183,32 @@ final class ReliableChannel implements AutoCloseable {
Consumer<PayloadOutputChannel> payloadWriter,
Function<PayloadInputChannel, T> payloadReader
) throws ClientException, ClientError {
- return applyOnDefaultChannel(channel -> channel.service(op, payloadWriter, payloadReader), op);
+ return service(op, payloadWriter, payloadReader, Collections.emptyList());
+ }
+
+ /**
+ * Send request to one of the passed nodes and handle response.
+ *
+ * @throws ClientException Thrown by {@code payloadWriter} or {@code payloadReader}.
+ * @throws ClientAuthenticationException When user name or password is invalid.
+ * @throws ClientAuthorizationException When user has no permission to perform operation.
+ * @throws ClientProtocolError When failed to handshake with server.
+ * @throws ClientServerError When failed to process request on server.
+ */
+ public <T> T service(
+ ClientOperation op,
+ Consumer<PayloadOutputChannel> payloadWriter,
+ Function<PayloadInputChannel, T> payloadReader,
+ List<UUID> targetNodes
+ ) throws ClientException, ClientError {
+ if (F.isEmpty(targetNodes))
+ return applyOnDefaultChannel(channel -> channel.service(op, payloadWriter, payloadReader), op);
+
+ return applyOnNodeChannelWithFallback(
+ targetNodes.get(ThreadLocalRandom.current().nextInt(targetNodes.size())),
+ channel -> channel.service(op, payloadWriter, payloadReader),
+ op
+ );
}
/**
@@ -943,7 +969,7 @@ final class ReliableChannel implements AutoCloseable {
/** Channel. */
private volatile ClientChannel ch;
- /** ID of the last server node that {@link ch} is or was connected to. */
+ /** ID of the last server node that {@link #ch} is or was connected to. */
private volatile UUID serverNodeId;
/** Address that holder is bind to (chCfg.addr) is not in use now. So close the holder. */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpIgniteClient.java b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpIgniteClient.java
index f15293089fb..a4d9922143f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpIgniteClient.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpIgniteClient.java
@@ -107,6 +107,9 @@ public class TcpIgniteClient implements IgniteClient {
/** Serializer/deserializer. */
private final ClientUtils serDes;
+ /** Logger. */
+ private final IgniteLogger log;
+
/**
* Private constructor. Use {@link TcpIgniteClient#start(ClientConfiguration)} to create an instance of
* {@code TcpIgniteClient}.
@@ -122,6 +125,8 @@ public class TcpIgniteClient implements IgniteClient {
BiFunction<ClientChannelConfiguration, ClientConnectionMultiplexer, ClientChannel> chFactory,
ClientConfiguration cfg
) throws ClientException {
+ log = NullLogger.whenNull(cfg.getLogger());
+
final ClientBinaryMetadataHandler metadataHnd = new ClientBinaryMetadataHandler();
ClientMarshallerContext marshCtx = new ClientMarshallerContext();
@@ -159,7 +164,7 @@ public class TcpIgniteClient implements IgniteClient {
compute = new ClientComputeImpl(ch, marsh, cluster.defaultClusterGroup());
- services = new ClientServicesImpl(ch, marsh, cluster.defaultClusterGroup());
+ services = new ClientServicesImpl(ch, marsh, cluster.defaultClusterGroup(), log);
lsnrsRegistry = new ClientCacheEntryListenersRegistry();
}
@@ -475,8 +480,6 @@ public class TcpIgniteClient implements IgniteClient {
if (clusterCfg == null)
return;
- IgniteLogger log = NullLogger.whenNull(cfg.getLogger());
-
if (log.isDebugEnabled())
log.debug("Cluster binary configuration retrieved: " + clusterCfg);
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientBitmaskFeature.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientBitmaskFeature.java
index e858bdbb8db..b0f673ac5ee 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientBitmaskFeature.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientBitmaskFeature.java
@@ -72,7 +72,10 @@ public enum ClientBitmaskFeature implements ThinProtocolFeature {
INDEX_QUERY(14),
/** IndexQuery limit. */
- INDEX_QUERY_LIMIT(15);
+ INDEX_QUERY_LIMIT(15),
+
+ /** Service topology. */
+ SERVICE_TOPOLOGY(16);
/** */
private static final EnumSet<ClientBitmaskFeature> ALL_FEATURES_AS_ENUM_SET =
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java
index eb765bdf5ed..13568187266 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientConnectionContext.java
@@ -78,10 +78,6 @@ public class ClientConnectionContext extends ClientListenerAbstractConnectionCon
/** Default version. */
public static final ClientListenerProtocolVersion DEFAULT_VER = VER_1_7_0;
- /** Default protocol context. */
- public static final ClientProtocolContext DEFAULT_PROTOCOL_CONTEXT =
- new ClientProtocolContext(DEFAULT_VER, ClientBitmaskFeature.allFeaturesAsEnumSet());
-
/** Supported versions. */
private static final Collection<ClientListenerProtocolVersion> SUPPORTED_VERS = Arrays.asList(
VER_1_7_0,
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientMessageParser.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientMessageParser.java
index 22cc257e660..ab3fe4c1608 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientMessageParser.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/ClientMessageParser.java
@@ -106,6 +106,7 @@ import org.apache.ignite.internal.processors.platform.client.datastructures.Clie
import org.apache.ignite.internal.processors.platform.client.service.ClientServiceGetDescriptorRequest;
import org.apache.ignite.internal.processors.platform.client.service.ClientServiceGetDescriptorsRequest;
import org.apache.ignite.internal.processors.platform.client.service.ClientServiceInvokeRequest;
+import org.apache.ignite.internal.processors.platform.client.service.ClientServiceTopologyRequest;
import org.apache.ignite.internal.processors.platform.client.streamer.ClientDataStreamerAddDataRequest;
import org.apache.ignite.internal.processors.platform.client.streamer.ClientDataStreamerStartRequest;
import org.apache.ignite.internal.processors.platform.client.tx.ClientTxEndRequest;
@@ -393,6 +394,9 @@ public class ClientMessageParser implements ClientListenerMessageParser {
/** IgniteSet.iterator page. */
private static final short OP_SET_ITERATOR_GET_PAGE = 9023;
+ /** Get service topology. */
+ private static final short OP_SERVICE_GET_TOPOLOGY = 7003;
+
/** Marshaller. */
private final GridBinaryMarshaller marsh;
@@ -697,6 +701,9 @@ public class ClientMessageParser implements ClientListenerMessageParser {
case OP_SET_ITERATOR_GET_PAGE:
return new ClientIgniteSetIteratorGetPageRequest(reader);
+
+ case OP_SERVICE_GET_TOPOLOGY:
+ return new ClientServiceTopologyRequest(reader);
}
return new ClientRawRequest(reader.readLong(), ClientStatus.INVALID_OP_CODE,
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceMappingsResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceMappingsResponse.java
new file mode 100644
index 00000000000..6be07fb886a
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceMappingsResponse.java
@@ -0,0 +1,55 @@
+/*
+ * 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.processors.platform.client.service;
+
+import java.util.Collection;
+import java.util.UUID;
+import org.apache.ignite.internal.binary.BinaryRawWriterEx;
+import org.apache.ignite.internal.client.thin.ClientUtils;
+import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext;
+import org.apache.ignite.internal.processors.platform.client.ClientResponse;
+import org.jetbrains.annotations.Nullable;
+
+/** Service topology response. */
+public class ClientServiceMappingsResponse extends ClientResponse {
+ /** Service instance nodes. */
+ @Nullable private final Collection<UUID> svcsNodes;
+
+ /**
+ * @param reqId Request id.
+ * @param svcsNodes Services instance nodes.
+ */
+ public ClientServiceMappingsResponse(long reqId, @Nullable Collection<UUID> svcsNodes) {
+ super(reqId);
+
+ this.svcsNodes = svcsNodes;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void encode(ClientConnectionContext ctx, BinaryRawWriterEx writer) {
+ super.encode(ctx, writer);
+
+ ClientUtils.collection(
+ svcsNodes,
+ writer.out(),
+ (out, uuid) -> {
+ out.writeLong(uuid.getMostSignificantBits());
+ out.writeLong(uuid.getLeastSignificantBits());
+ });
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceTopologyRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceTopologyRequest.java
new file mode 100644
index 00000000000..560c79c0560
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/platform/client/service/ClientServiceTopologyRequest.java
@@ -0,0 +1,63 @@
+/*
+ * 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.processors.platform.client.service;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.UUID;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.binary.BinaryRawReader;
+import org.apache.ignite.internal.processors.platform.client.ClientConnectionContext;
+import org.apache.ignite.internal.processors.platform.client.ClientRequest;
+import org.apache.ignite.internal.processors.platform.client.ClientResponse;
+import org.apache.ignite.internal.processors.platform.client.ClientStatus;
+import org.apache.ignite.internal.processors.platform.client.IgniteClientException;
+import org.apache.ignite.internal.util.typedef.F;
+
+/**
+ * Request topology of certain service.
+ */
+public class ClientServiceTopologyRequest extends ClientRequest {
+ /** The service name. */
+ private final String name;
+
+ /**
+ * Creates the service topology request.
+ *
+ * @param reader Reader to read the {@link #name} from.
+ */
+ public ClientServiceTopologyRequest(BinaryRawReader reader) {
+ super(reader);
+
+ name = reader.readString();
+ }
+
+ /** {@inheritDoc} */
+ @Override public ClientResponse process(ClientConnectionContext ctx) {
+ Map<UUID, Integer> srvcTop;
+
+ try {
+ srvcTop = ctx.kernalContext().service().serviceTopology(name, 0);
+ }
+ catch (IgniteCheckedException e) {
+ throw new IgniteClientException(ClientStatus.FAILED, "Failed to get topology for service '" + name + "'.", e);
+ }
+
+ return new ClientServiceMappingsResponse(requestId(), F.isEmpty(srvcTop) ? Collections.emptyList() : srvcTop.keySet());
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java
index 598c3e97df0..dcceab41d9d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java
@@ -936,7 +936,7 @@ public class IgniteServiceProcessor extends GridProcessorAdapter implements Igni
* @return Service topology.
* @throws IgniteCheckedException On error.
*/
- public Map<UUID, Integer> serviceTopology(String name, long timeout) throws IgniteCheckedException {
+ @Nullable public Map<UUID, Integer> serviceTopology(String name, long timeout) throws IgniteCheckedException {
assert timeout >= 0;
long startTime = U.currentTimeMillis();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentTask.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentTask.java
index 37828dd2474..d644f90c281 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentTask.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentTask.java
@@ -132,7 +132,7 @@ class ServiceDeploymentTask {
this.depId = depId;
this.ctx = ctx;
- srvcProc = (IgniteServiceProcessor)ctx.service();
+ srvcProc = ctx.service();
log = ctx.log(getClass());
}
diff --git a/modules/core/src/test/java/org/apache/ignite/client/ReliabilityTest.java b/modules/core/src/test/java/org/apache/ignite/client/ReliabilityTest.java
index 5a014365a15..b7ea340f36c 100644
--- a/modules/core/src/test/java/org/apache/ignite/client/ReliabilityTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/client/ReliabilityTest.java
@@ -380,7 +380,7 @@ public class ReliabilityTest extends AbstractThinClientTest {
String nullOpsNames = nullOps.stream().map(Enum::name).collect(Collectors.joining(", "));
- long expectedNullCnt = 21;
+ long expectedNullCnt = 22;
String msg = nullOps.size()
+ " operation codes do not have public equivalent. When adding new codes, update ClientOperationType too. Missing ops: "
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/client/thin/ServiceAwarenessTest.java b/modules/core/src/test/java/org/apache/ignite/internal/client/thin/ServiceAwarenessTest.java
new file mode 100644
index 00000000000..94fba7293f3
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/client/thin/ServiceAwarenessTest.java
@@ -0,0 +1,622 @@
+/*
+ * 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.client.thin;
+
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.client.ClientAuthenticationException;
+import org.apache.ignite.client.ClientConnectionException;
+import org.apache.ignite.client.ClientException;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.ClientConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.GridJobExecuteRequest;
+import org.apache.ignite.internal.GridTopic;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.client.thin.io.ClientConnectionMultiplexer;
+import org.apache.ignite.internal.managers.communication.GridMessageListener;
+import org.apache.ignite.internal.managers.discovery.CustomMessageWrapper;
+import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.internal.processors.service.GridServiceProxy;
+import org.apache.ignite.internal.processors.service.ServiceClusterDeploymentResultBatch;
+import org.apache.ignite.internal.util.GridConcurrentHashSet;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.services.ServiceConfiguration;
+import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.testframework.ListeningTestLogger;
+import org.apache.ignite.testframework.junits.logger.GridTestLog4jLogger;
+import org.apache.logging.log4j.Level;
+import org.jetbrains.annotations.Nullable;
+import org.junit.Test;
+
+import static org.apache.ignite.testframework.GridTestUtils.runAsync;
+import static org.apache.ignite.testframework.GridTestUtils.runMultiThreadedAsync;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
+/**
+ * Checks the service awareness feature of the thin client.
+ */
+public class ServiceAwarenessTest extends AbstractThinClientTest {
+ /** Node-filter service name. */
+ private static final String SRV_NAME = "node_filtered_svc";
+
+ /** Number of grids at the test start. */
+ private static final int GRIDS = 4;
+
+ /** Number of node instances with the initial service deployment. */
+ private static final int INIT_SRVC_NODES_CNT = 2;
+
+ /** */
+ protected boolean partitionAwareness = true;
+
+ /** */
+ private static ListeningTestLogger clientLogLsnr;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ cfg.setDiscoverySpi(new TestBlockingDiscoverySpi());
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected ClientConfiguration getClientConfiguration() {
+ ClientConfiguration ccfg = super.getClientConfiguration();
+
+ ccfg.setLogger(clientLogLsnr);
+
+ return ccfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ clientLogLsnr = new ListeningTestLogger(log);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected long getTestTimeout() {
+ return 45_000L;
+ }
+
+ /** */
+ private static ServiceConfiguration serviceCfg() {
+ // Service is deployed on nodes with the name index equal to 1, 2 or >= GRIDS.
+ return new ServiceConfiguration()
+ .setName(SRV_NAME)
+ .setService(new ServicesTest.TestService())
+ .setMaxPerNodeCount(1)
+ .setNodeFilter(new TestNodeFilter());
+ }
+
+ /** {@inheritDoc} */
+ @Override protected boolean isClientPartitionAwarenessEnabled() {
+ return partitionAwareness;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ ((GridTestLog4jLogger)log).setLevel(Level.INFO);
+
+ stopAllGrids();
+
+ partitionAwareness = true;
+
+ clientLogLsnr.clearListeners();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ startGrids(GRIDS);
+
+ grid(1).services().deploy(serviceCfg());
+ }
+
+ /** */
+ @Test
+ public void testDelayedServiceRedeploy() throws Exception {
+ TestBlockingDiscoverySpi testDisco = ((TestBlockingDiscoverySpi)grid(0).configuration().getDiscoverySpi());
+
+ // Service topology on the client.
+ Set<UUID> srvcTopOnClient = new GridConcurrentHashSet<>();
+
+ addSrvcTopUpdateClientLogLsnr(srvcTopOnClient::addAll);
+
+ AtomicBoolean svcRunFlag = new AtomicBoolean(true);
+
+ try (IgniteClient client = startClient()) {
+ ServicesTest.TestServiceInterface svc = client.services().serviceProxy(SRV_NAME, ServicesTest.TestServiceInterface.class);
+
+ runAsync(() -> {
+ while (svcRunFlag.get())
+ svc.testMethod();
+ });
+
+ waitForCondition(() -> srvcTopOnClient.size() == INIT_SRVC_NODES_CNT
+ && srvcTopOnClient.contains(grid(1).localNode().id())
+ && srvcTopOnClient.contains(grid(2).localNode().id()),
+ getTestTimeout());
+
+ // Delays service redeployment and the service topology update on the server side.
+ testDisco.toBlock.add(ServiceClusterDeploymentResultBatch.class);
+
+ startGrid(GRIDS);
+
+ waitForCondition(() -> testDisco.blocked.size() == 1, getTestTimeout());
+
+ // Ensure all the nodes have started but the service topology hasn't updated yet.
+ for (Ignite ig : G.allGrids()) {
+ assertEquals(ig.cluster().nodes().size(), GRIDS + 1);
+
+ // Ensure there are still SRVC_FILTERED_NOIDES_CNT nodes with the service instance.
+ assertEquals(((IgniteEx)ig).context().service().serviceTopology(SRV_NAME, 0).size(),
+ INIT_SRVC_NODES_CNT);
+ }
+
+ // Ensure the client's topology is not updated.
+ assertTrue(srvcTopOnClient.size() == INIT_SRVC_NODES_CNT
+ && !srvcTopOnClient.contains(grid(GRIDS).localNode().id()));
+
+ testDisco.release();
+
+ // Ensure the service topology has been updated to 3 instances per cluster.
+ for (Ignite ig : G.allGrids()) {
+ waitForCondition(
+ () -> {
+ try {
+ return ((IgniteEx)ig).context().service().serviceTopology(SRV_NAME, 0).size() == 3;
+ }
+ catch (Exception e) {
+ return false;
+ }
+ },
+ getTestTimeout()
+ );
+ }
+
+ waitForCondition(() -> srvcTopOnClient.size() == 3 && srvcTopOnClient.contains(grid(1).localNode().id())
+ && srvcTopOnClient.contains(grid(2).localNode().id())
+ && srvcTopOnClient.contains(grid(GRIDS).localNode().id()), getTestTimeout());
+ }
+ finally {
+ svcRunFlag.set(false);
+ }
+ }
+
+ /**
+ * Tests several nodes come while one thread is used to call the service.
+ */
+ @Test
+ public void testNodesJoinSingleThreaded() throws Exception {
+ doTestClusterTopChangesWhileServiceCalling(3, true, false);
+ }
+
+ /**
+ * Tests several nodes come while several threads are used to call the service.
+ */
+ @Test
+ public void testNodesJoinMultiThreaded() throws Exception {
+ doTestClusterTopChangesWhileServiceCalling(3, true, true);
+ }
+
+ /**
+ * Tests several nodes leaves while one thread is used to call the service.
+ */
+ @Test
+ public void testNodesLeaveSingleThreaded() throws Exception {
+ doTestClusterTopChangesWhileServiceCalling(3, false, false);
+ }
+
+ /**
+ * Tests several nodes leave while several threads are used to call the service.
+ */
+ @Test
+ public void testNodesLeaveMultiThreaded() throws Exception {
+ doTestClusterTopChangesWhileServiceCalling(3, false, true);
+ }
+
+ /**
+ * Tests change of the minor cluster topology version doesn't trigger the service topology update.
+ */
+ @Test
+ public void testMinorTopologyVersionDoesntAffect() throws Exception {
+ try (IgniteClient client = startClient()) {
+ ServicesTest.TestServiceInterface svc = client.services().serviceProxy(SRV_NAME, ServicesTest.TestServiceInterface.class);
+
+ Set<UUID> srvcTopOnClient = new GridConcurrentHashSet<>();
+
+ addSrvcTopUpdateClientLogLsnr(srvcTopOnClient::addAll);
+
+ ((GridTestLog4jLogger)log).setLevel(Level.DEBUG);
+
+ while (srvcTopOnClient.isEmpty())
+ svc.testMethod();
+
+ // Last time ot the topology update.
+ long time = System.nanoTime();
+
+ srvcTopOnClient.clear();
+
+ AffinityTopologyVersion prevTopVersion = grid(0).context().discovery().topologyVersionEx();
+
+ grid(0).createCache("testCache");
+
+ awaitPartitionMapExchange();
+
+ AffinityTopologyVersion newTopVersion = grid(0).context().discovery().topologyVersionEx();
+
+ assertTrue(newTopVersion.topologyVersion() == prevTopVersion.topologyVersion()
+ && newTopVersion.minorTopologyVersion() > prevTopVersion.minorTopologyVersion());
+
+ while (srvcTopOnClient.isEmpty())
+ svc.testMethod();
+
+ // Update only by the timeout.
+ assertTrue(U.nanosToMillis(System.nanoTime() - time) > ClientServicesImpl.SRV_TOP_UPDATE_PERIOD / 2);
+ }
+ }
+
+ /**
+ * Tests the service topology update with a gap of service invocation during forced service redeployment.
+ */
+ @Test
+ public void testForcedServiceRedeployWhileClientIsIdle() {
+ try (IgniteClient client = startClient()) {
+ ServicesTest.TestServiceInterface svc = client.services().serviceProxy(SRV_NAME, ServicesTest.TestServiceInterface.class);
+
+ Set<UUID> srvcTopOnClient = new GridConcurrentHashSet<>();
+
+ addSrvcTopUpdateClientLogLsnr(srvcTopOnClient::addAll);
+
+ ((GridTestLog4jLogger)log).setLevel(Level.DEBUG);
+
+ while (srvcTopOnClient.isEmpty())
+ svc.testMethod();
+
+ ((GridTestLog4jLogger)log).setLevel(Level.INFO);
+
+ assertTrue(srvcTopOnClient.size() == INIT_SRVC_NODES_CNT
+ && srvcTopOnClient.contains(grid(1).localNode().id())
+ && srvcTopOnClient.contains(grid(2).localNode().id()));
+
+ long prevTopVersion = grid(0).context().discovery().topologyVersion();
+
+ grid(1).services().cancel(SRV_NAME);
+
+ srvcTopOnClient.clear();
+
+ grid(1).services().deploy(serviceCfg().setNodeFilter(null));
+
+ assertEquals(prevTopVersion, grid(0).context().discovery().topologyVersion());
+
+ ((GridTestLog4jLogger)log).setLevel(Level.DEBUG);
+
+ while (srvcTopOnClient.isEmpty())
+ svc.testMethod();
+
+ assertEquals(GRIDS, srvcTopOnClient.size());
+
+ for (Ignite ig : G.allGrids())
+ assertTrue(srvcTopOnClient.contains(ig.cluster().localNode().id()));
+ }
+ }
+
+ /** */
+ private void doTestClusterTopChangesWhileServiceCalling(
+ int nodesCnt,
+ boolean addNodes,
+ boolean multiThreaded)
+ throws Exception {
+ assert nodesCnt > 0;
+
+ Set<UUID> newNodesUUIDs = new GridConcurrentHashSet<>();
+
+ // Start additional nodes to stop them.
+ if (!addNodes) {
+ startGridsMultiThreaded(GRIDS, nodesCnt);
+
+ for (int i = GRIDS; i < GRIDS + nodesCnt; ++i)
+ newNodesUUIDs.add(grid(i).localNode().id());
+ }
+
+ // Service topology on the clients.
+ Set<UUID> srvcTopOnClient = new GridConcurrentHashSet<>();
+
+ addSrvcTopUpdateClientLogLsnr(srvcTopOnClient::addAll);
+
+ AtomicBoolean changeClusterTop = new AtomicBoolean();
+ AtomicBoolean stopFlag = new AtomicBoolean();
+
+ try (IgniteClient client = startClient()) {
+ ServicesTest.TestServiceInterface svc = client.services().serviceProxy(SRV_NAME, ServicesTest.TestServiceInterface.class);
+
+ ((GridTestLog4jLogger)log).setLevel(Level.DEBUG);
+
+ IgniteInternalFuture<?> runFut = runMultiThreadedAsync(() -> {
+ do {
+ try {
+ svc.testMethod();
+ }
+ catch (ClientException e) {
+ String m = e.getMessage();
+
+ // TODO: IGNITE-20802 : Exception should not occur.
+ // Client doesn't retry service invocation if the redirected-to service instance node leaves cluster.
+ if (addNodes || (!m.contains("Node has left grid") && !m.contains("Failed to send job due to node failure"))
+ || newNodesUUIDs.stream().noneMatch(nid -> m.contains(nid.toString())))
+ throw e;
+ }
+ }
+ while (!stopFlag.get());
+ }, multiThreaded ? 4 : 1, "ServiceTestLoader");
+
+ while (!stopFlag.get()) {
+ // Wait until the initial topology is received.
+ if (srvcTopOnClient.size() == (addNodes ? INIT_SRVC_NODES_CNT : INIT_SRVC_NODES_CNT + nodesCnt)
+ && changeClusterTop.compareAndSet(false, true)) {
+ srvcTopOnClient.clear();
+
+ for (int i = 0; i < nodesCnt; ++i) {
+ int nodeIdx = GRIDS + i;
+
+ runAsync(() -> {
+ try {
+ if (addNodes)
+ newNodesUUIDs.add(startGrid(nodeIdx).localNode().id());
+ else
+ stopGrid(nodeIdx);
+ }
+ catch (Exception e) {
+ log.error("Unable to start or stop test grid.", e);
+
+ stopFlag.set(true);
+ }
+ });
+ }
+ }
+
+ // Stop if new excepted service topology received.
+ if (srvcTopOnClient.size() == (addNodes ? INIT_SRVC_NODES_CNT + nodesCnt : INIT_SRVC_NODES_CNT))
+ stopFlag.set(true);
+
+ Thread.sleep(10);
+ }
+
+ runFut.get();
+ }
+
+ // The initial nodes must always persist it the service topology.
+ assertTrue(srvcTopOnClient.contains(grid(1).localNode().id())
+ && srvcTopOnClient.contains(grid(2).localNode().id()));
+
+ assertEquals(addNodes ? nodesCnt : 0, newNodesUUIDs.stream().filter(srvcTopOnClient::contains).count());
+ }
+
+ /**
+ * Tests client uses service awareness when partitionAwareness is enabled.
+ */
+ @Test
+ public void testServiceAwarenessEnabled() {
+ // Counters of the invocation redirects.
+ AtomicInteger redirectCnt = new AtomicInteger();
+
+ // Service topology received by the client.
+ Set<UUID> top = new GridConcurrentHashSet<>();
+
+ // Requested server nodes with a service invocation.
+ Collection<UUID> requestedServers = new GridConcurrentHashSet<>();
+
+ addSrvcTopUpdateClientLogLsnr(uuids -> {
+ // Reset counters on the first topology update.
+ if (top.isEmpty())
+ redirectCnt.set(0);
+
+ top.addAll(uuids);
+ });
+
+ // Listener of the service remote call (the redirection).
+ G.allGrids().forEach(g -> ((IgniteEx)g).context().io().addMessageListener(GridTopic.TOPIC_JOB, new GridMessageListener() {
+ @Override public void onMessage(UUID nodeId, Object msg, byte plc) {
+ if (msg instanceof GridJobExecuteRequest
+ && ((GridJobExecuteRequest)msg).getTaskClassName().contains(GridServiceProxy.class.getName()))
+ redirectCnt.incrementAndGet();
+ }
+ }));
+
+ partitionAwareness = false;
+
+ ((GridTestLog4jLogger)log).setLevel(Level.DEBUG);
+
+ try (IgniteClient client = startClient(requestedServers)) {
+ ServicesTest.TestServiceInterface svc = client.services().serviceProxy(SRV_NAME, ServicesTest.TestServiceInterface.class);
+
+ for (int i = 0; i < 100; ++i)
+ svc.testMethod();
+ }
+
+ // Check no service awareness: continous redirections.
+ assertEquals(100, redirectCnt.get());
+
+ // Ensure that client received no service topology update.
+ assertTrue(top.isEmpty());
+
+ assertTrue(requestedServers.size() == 1 && requestedServers.contains(grid(0).localNode().id()));
+
+ partitionAwareness = true;
+
+ try (IgniteClient client = startClient(requestedServers)) {
+ ServicesTest.TestServiceInterface svc = client.services().serviceProxy(SRV_NAME, ServicesTest.TestServiceInterface.class);
+
+ // We assume that the topology will be received and used for the further requests.
+ for (int i = 0; i < 1000; ++i)
+ svc.testMethod();
+
+ redirectCnt.set(0);
+ requestedServers.clear();
+
+ for (int i = 0; i < 1000; ++i)
+ svc.testMethod();
+ }
+
+ // Check the received topology.
+ assertTrue(top.size() == INIT_SRVC_NODES_CNT && top.contains(grid(1).localNode().id())
+ && top.contains(grid(2).localNode().id()));
+
+ // Ensure that only the target nodes were requeted after the topology getting.
+ assertEquals(top, requestedServers);
+
+ // Ensure there were no redirected sertvic calls any more.
+ assertEquals(0, redirectCnt.get());
+ }
+
+ /** */
+ private static void addSrvcTopUpdateClientLogLsnr(Consumer<Set<UUID>> srvTopConsumer) {
+ clientLogLsnr.registerListener(s -> {
+ if (s.contains("Topology of service '" + SRV_NAME + "' has been updated. The service instance nodes: ")) {
+ String nodes = s.substring(s.indexOf(": [") + 3);
+
+ nodes = nodes.substring(0, nodes.length() - 1);
+
+ srvTopConsumer.accept(Arrays.stream(nodes.split(", ")).map(UUID::fromString).collect(Collectors.toSet()));
+ }
+ });
+ }
+
+ /** */
+ private IgniteClient startClient() {
+ return new TcpIgniteClient((cfg, hnd) -> new TestTcpChannel(cfg, hnd, null),
+ getClientConfiguration(grid(0)));
+ }
+
+ /** */
+ private IgniteClient startClient(@Nullable Collection<UUID> requestedServerNodes) {
+ return new TcpIgniteClient((cfg, hnd) -> new TestTcpChannel(cfg, hnd, requestedServerNodes),
+ getClientConfiguration(grid(0)));
+ }
+
+ /**
+ * Accepts nodes with the name index equal to 1, 2 or >= GRIDS.
+ */
+ private static final class TestNodeFilter implements IgnitePredicate<ClusterNode> {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** {@inheritDoc} */
+ @Override public boolean apply(ClusterNode node) {
+ String nodeName = node.attribute("org.apache.ignite.ignite.name");
+
+ if (F.isEmpty(nodeName))
+ return false;
+
+ int nodeIdx = -1;
+
+ try {
+ nodeIdx = Integer.parseInt(nodeName.substring(nodeName.length() - 1));
+ }
+ catch (Exception e) {
+ // No-op.
+ }
+
+ return nodeIdx == 1 || nodeIdx == 2 || nodeIdx >= GRIDS;
+ }
+ }
+
+ /** */
+ private static final class TestBlockingDiscoverySpi extends TcpDiscoverySpi {
+ /** */
+ private final Set<Class<? extends DiscoveryCustomMessage>> toBlock = new HashSet<>();
+
+ /** */
+ private final List<CustomMessageWrapper> blocked = new CopyOnWriteArrayList<>();
+
+ /** {@inheritDoc} */
+ @Override public void sendCustomEvent(DiscoverySpiCustomMessage msg) throws IgniteException {
+ if (msg instanceof CustomMessageWrapper
+ && toBlock.stream().anyMatch(mt -> mt.isAssignableFrom(((CustomMessageWrapper)msg).delegate().getClass()))) {
+ blocked.add((CustomMessageWrapper)msg);
+
+ return;
+ }
+
+ super.sendCustomEvent(msg);
+ }
+
+ /** */
+ public void release() {
+ toBlock.clear();
+
+ blocked.forEach(this::sendCustomEvent);
+ }
+ }
+
+ /**
+ * A client connection channel able to register the server nodes requested to call a service.
+ */
+ private static final class TestTcpChannel extends TcpClientChannel {
+ /** */
+ private final @Nullable Collection<UUID> requestedServerNodes;
+
+ /** Ctor. */
+ private TestTcpChannel(
+ ClientChannelConfiguration cfg,
+ ClientConnectionMultiplexer connMgr,
+ @Nullable Collection<UUID> requestedServerNodes)
+ throws ClientConnectionException, ClientAuthenticationException, ClientProtocolError {
+ super(cfg, connMgr);
+
+ this.requestedServerNodes = requestedServerNodes;
+ }
+
+ /** {@inheritDoc} */
+ @Override public <T> T service(ClientOperation op, Consumer<PayloadOutputChannel> payloadWriter,
+ Function<PayloadInputChannel, T> payloadReader) throws ClientException {
+ UUID srvNodeId = serverNodeId();
+
+ if (op == ClientOperation.SERVICE_INVOKE && requestedServerNodes != null && srvNodeId != null)
+ requestedServerNodes.add(srvNodeId);
+
+ return super.service(op, payloadWriter, payloadReader);
+ }
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
index 3069a56313f..f9510ae8f36 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/future/GridFutureAdapterSelfTest.java
@@ -444,7 +444,7 @@ public class GridFutureAdapterSelfTest extends GridCommonAbstractTest {
}
/**
- * Completes the future exceptionally, if {@code res} is the instanse of {@link Throwable}.
+ * Completes the future exceptionally, if {@code res} is the instance of {@link Throwable}.
*
* @param fut Future.
* @param res Result.
diff --git a/modules/indexing/src/test/java/org/apache/ignite/client/ClientTestSuite.java b/modules/indexing/src/test/java/org/apache/ignite/client/ClientTestSuite.java
index 04d9e2b76d1..588715be857 100644
--- a/modules/indexing/src/test/java/org/apache/ignite/client/ClientTestSuite.java
+++ b/modules/indexing/src/test/java/org/apache/ignite/client/ClientTestSuite.java
@@ -31,6 +31,7 @@ import org.apache.ignite.internal.client.thin.InactiveClusterCacheRequestTest;
import org.apache.ignite.internal.client.thin.MetadataRegistrationTest;
import org.apache.ignite.internal.client.thin.OptimizedMarshallerClassesCachedTest;
import org.apache.ignite.internal.client.thin.ReliableChannelTest;
+import org.apache.ignite.internal.client.thin.ServiceAwarenessTest;
import org.apache.ignite.internal.client.thin.ServicesBinaryArraysTests;
import org.apache.ignite.internal.client.thin.ServicesTest;
import org.apache.ignite.internal.client.thin.ThinClientEnpointsDiscoveryTest;
@@ -68,6 +69,7 @@ import org.junit.runners.Suite;
ClusterGroupTest.class,
ServicesTest.class,
ServicesBinaryArraysTests.class,
+ ServiceAwarenessTest.class,
CacheEntryListenersTest.class,
ThinClientPartitionAwarenessStableTopologyTest.class,
ThinClientPartitionAwarenessUnstableTopologyTest.class,