This is an automated email from the ASF dual-hosted git repository.
ibessonov 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 3062df341f IGNITE-18063 Use consistent IDs in assignments (#1354)
3062df341f is described below
commit 3062df341f2810d6bce80a618ac2e485b4e4337c
Author: Alexander Polovtcev <alex.polovt...@gmail.com>
AuthorDate: Fri Nov 18 11:55:10 2022 +0300
IGNITE-18063 Use consistent IDs in assignments (#1354)
---
.../internal/sql/engine/exec/ExchangeService.java | 44 +++++-----
.../sql/engine/exec/ExchangeServiceImpl.java | 52 ++++++------
.../internal/sql/engine/exec/ExecutionContext.java | 30 +++----
.../sql/engine/exec/ExecutionServiceImpl.java | 61 +++++++-------
.../sql/engine/exec/LogicalRelImplementor.java | 12 +--
.../sql/engine/exec/MailboxRegistryImpl.java | 4 +-
.../sql/engine/exec/RemoteFragmentKey.java | 18 ++--
.../ignite/internal/sql/engine/exec/rel/Inbox.java | 80 +++++++++---------
.../internal/sql/engine/exec/rel/Outbox.java | 55 +++++++------
.../sql/engine/message/MessageListener.java | 6 +-
.../sql/engine/message/MessageService.java | 13 ++-
.../sql/engine/message/MessageServiceImpl.java | 40 ++++-----
.../sql/engine/metadata/ColocationGroup.java | 56 ++++++-------
.../sql/engine/metadata/FragmentDescription.java | 4 +-
.../sql/engine/metadata/FragmentMapping.java | 10 +--
.../engine/metadata/IgniteMdFragmentMapping.java | 2 +-
.../sql/engine/metadata/MappingServiceImpl.java | 2 +-
.../sql/engine/metadata/RemoteException.java | 18 ++--
.../sql/engine/prepare/AbstractMultiStepPlan.java | 2 +-
.../internal/sql/engine/prepare/Fragment.java | 8 +-
.../sql/engine/prepare/MappingQueryContext.java | 12 +--
.../sql/engine/trait/DistributionFunction.java | 12 +--
.../internal/sql/engine/StopCalciteModuleTest.java | 33 +++-----
.../sql/engine/exec/ExecutionServiceImplTest.java | 95 +++++++++++-----------
.../exec/rel/TableScanNodeExecutionTest.java | 7 +-
.../ignite/distributed/ItTablePersistenceTest.java | 3 -
.../distributed/ItTxDistributedTestSingleNode.java | 2 -
.../ignite/internal/table/ItColocationTest.java | 3 +-
.../internal/table/distributed/TableManager.java | 25 ++----
.../distributed/storage/InternalTableImpl.java | 9 +-
.../table/impl/DummyInternalTableImpl.java | 3 +-
31 files changed, 342 insertions(+), 379 deletions(-)
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeService.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeService.java
index f0d9eaffee..3724784dae 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeService.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeService.java
@@ -29,64 +29,64 @@ public interface ExchangeService extends LifecycleAware {
/**
* Sends a batch of data to remote node.
*
- * @param nodeId Target node ID.
- * @param qryId Query ID.
+ * @param nodeName Target node consistent ID.
+ * @param qryId Query ID.
* @param fragmentId Target fragment ID.
* @param exchangeId Exchange ID.
- * @param batchId Batch ID.
- * @param last Last batch flag.
- * @param rows Data rows.
+ * @param batchId Batch ID.
+ * @param last Last batch flag.
+ * @param rows Data rows.
*/
- <RowT> void sendBatch(String nodeId, UUID qryId, long fragmentId, long
exchangeId, int batchId, boolean last,
+ <RowT> void sendBatch(String nodeName, UUID qryId, long fragmentId, long
exchangeId, int batchId, boolean last,
List<RowT> rows) throws IgniteInternalCheckedException;
/**
* Acknowledges a batch with given ID is processed.
*
- * @param nodeId Node ID to notify.
- * @param qryId Query ID.
+ * @param nodeName Node consistent ID to notify.
+ * @param qryId Query ID.
* @param fragmentId Target fragment ID.
* @param exchangeId Exchange ID.
- * @param batchId Batch ID.
+ * @param batchId Batch ID.
*/
- void acknowledge(String nodeId, UUID qryId, long fragmentId, long
exchangeId, int batchId) throws IgniteInternalCheckedException;
+ void acknowledge(String nodeName, UUID qryId, long fragmentId, long
exchangeId, int batchId) throws IgniteInternalCheckedException;
/**
* Sends cancel request.
*
- * @param nodeId Target node ID.
- * @param qryId Query ID.
+ * @param nodeName Target node consistent ID.
+ * @param qryId Query ID.
* @param fragmentId Target fragment ID.
* @param exchangeId Exchange ID.
*/
- void closeInbox(String nodeId, UUID qryId, long fragmentId, long
exchangeId) throws IgniteInternalCheckedException;
+ void closeInbox(String nodeName, UUID qryId, long fragmentId, long
exchangeId) throws IgniteInternalCheckedException;
/**
* Sends cancel request.
*
- * @param nodeId Target node ID.
- * @param qryId Query ID.
+ * @param nodeName Target node consistent ID.
+ * @param qryId Query ID.
*/
- void closeQuery(String nodeId, UUID qryId) throws
IgniteInternalCheckedException;
+ void closeQuery(String nodeName, UUID qryId) throws
IgniteInternalCheckedException;
/**
* Send error.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*
- * @param nodeId Target node ID.
- * @param qryId Query ID.
+ * @param nodeName Target node consistent ID.
+ * @param qryId Query ID.
* @param fragmentId Source fragment ID.
- * @param err Exception to send.
+ * @param err Exception to send.
* @throws IgniteInternalCheckedException On error marshaling or send
ErrorMessage.
*/
- void sendError(String nodeId, UUID qryId, long fragmentId, Throwable err)
throws IgniteInternalCheckedException;
+ void sendError(String nodeName, UUID qryId, long fragmentId, Throwable
err) throws IgniteInternalCheckedException;
/**
* Alive.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*
- * @param nodeId Node ID.
+ * @param nodeName Target node consistent ID.
* @return {@code true} if node is alive, {@code false} otherwise.
*/
- boolean alive(String nodeId);
+ boolean alive(String nodeName);
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeServiceImpl.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeServiceImpl.java
index 73d94a4ef0..5f26aa2585 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeServiceImpl.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExchangeServiceImpl.java
@@ -83,10 +83,10 @@ public class ExchangeServiceImpl implements ExchangeService
{
/** {@inheritDoc} */
@Override
- public <RowT> void sendBatch(String nodeId, UUID qryId, long fragmentId,
long exchangeId, int batchId,
+ public <RowT> void sendBatch(String nodeName, UUID qryId, long fragmentId,
long exchangeId, int batchId,
boolean last, List<RowT> rows) throws
IgniteInternalCheckedException {
msgSrvc.send(
- nodeId,
+ nodeName,
FACTORY.queryBatchMessage()
.queryId(qryId)
.fragmentId(fragmentId)
@@ -100,10 +100,10 @@ public class ExchangeServiceImpl implements
ExchangeService {
/** {@inheritDoc} */
@Override
- public void acknowledge(String nodeId, UUID qryId, long fragmentId, long
exchangeId, int batchId)
+ public void acknowledge(String nodeName, UUID qryId, long fragmentId, long
exchangeId, int batchId)
throws IgniteInternalCheckedException {
msgSrvc.send(
- nodeId,
+ nodeName,
FACTORY.queryBatchAcknowledgeMessage()
.queryId(qryId)
.fragmentId(fragmentId)
@@ -115,9 +115,9 @@ public class ExchangeServiceImpl implements ExchangeService
{
/** {@inheritDoc} */
@Override
- public void closeQuery(String nodeId, UUID qryId) throws
IgniteInternalCheckedException {
+ public void closeQuery(String nodeName, UUID qryId) throws
IgniteInternalCheckedException {
msgSrvc.send(
- nodeId,
+ nodeName,
FACTORY.queryCloseMessage()
.queryId(qryId)
.build()
@@ -126,9 +126,9 @@ public class ExchangeServiceImpl implements ExchangeService
{
/** {@inheritDoc} */
@Override
- public void closeInbox(String nodeId, UUID qryId, long fragmentId, long
exchangeId) throws IgniteInternalCheckedException {
+ public void closeInbox(String nodeName, UUID qryId, long fragmentId, long
exchangeId) throws IgniteInternalCheckedException {
msgSrvc.send(
- nodeId,
+ nodeName,
FACTORY.inboxCloseMessage()
.queryId(qryId)
.fragmentId(fragmentId)
@@ -139,9 +139,9 @@ public class ExchangeServiceImpl implements ExchangeService
{
/** {@inheritDoc} */
@Override
- public void sendError(String nodeId, UUID qryId, long fragmentId,
Throwable err) throws IgniteInternalCheckedException {
+ public void sendError(String nodeName, UUID qryId, long fragmentId,
Throwable err) throws IgniteInternalCheckedException {
msgSrvc.send(
- nodeId,
+ nodeName,
FACTORY.errorMessage()
.queryId(qryId)
.fragmentId(fragmentId)
@@ -152,11 +152,11 @@ public class ExchangeServiceImpl implements
ExchangeService {
/** {@inheritDoc} */
@Override
- public boolean alive(String nodeId) {
- return msgSrvc.alive(nodeId);
+ public boolean alive(String nodeName) {
+ return msgSrvc.alive(nodeName);
}
- protected void onMessage(String nodeId, InboxCloseMessage msg) {
+ private void onMessage(String nodeName, InboxCloseMessage msg) {
Collection<Inbox<?>> inboxes = mailboxRegistry.inboxes(msg.queryId(),
msg.fragmentId(), msg.exchangeId());
if (!nullOrEmpty(inboxes)) {
@@ -164,35 +164,35 @@ public class ExchangeServiceImpl implements
ExchangeService {
inbox.context().execute(inbox::close, inbox::onError);
}
} else if (LOG.isDebugEnabled()) {
- LOG.debug("Stale inbox cancel message received [nodeId={},
queryId={}, fragmentId={}, exchangeId={}]",
- nodeId, msg.queryId(), msg.fragmentId(), msg.exchangeId());
+ LOG.debug("Stale inbox cancel message received [nodeName={},
queryId={}, fragmentId={}, exchangeId={}]",
+ nodeName, msg.queryId(), msg.fragmentId(),
msg.exchangeId());
}
}
- protected void onMessage(String nodeId, QueryBatchAcknowledgeMessage msg) {
+ private void onMessage(String nodeName, QueryBatchAcknowledgeMessage msg) {
Outbox<?> outbox = mailboxRegistry.outbox(msg.queryId(),
msg.exchangeId());
if (outbox != null) {
try {
- outbox.onAcknowledge(nodeId, msg.batchId());
+ outbox.onAcknowledge(nodeName, msg.batchId());
} catch (Throwable e) {
outbox.onError(e);
throw new IgniteInternalException(UNEXPECTED_ERR, "Unexpected
exception", e);
}
} else if (LOG.isDebugEnabled()) {
- LOG.debug("Stale acknowledge message received: [nodeId={},
queryId={}, fragmentId={}, exchangeId={}, batchId={}]",
- nodeId, msg.queryId(), msg.fragmentId(), msg.exchangeId(),
msg.batchId());
+ LOG.debug("Stale acknowledge message received: [nodeName={},
queryId={}, fragmentId={}, exchangeId={}, batchId={}]",
+ nodeName, msg.queryId(), msg.fragmentId(),
msg.exchangeId(), msg.batchId());
}
}
- protected void onMessage(String nodeId, QueryBatchMessage msg) {
+ private void onMessage(String nodeName, QueryBatchMessage msg) {
Inbox<?> inbox = mailboxRegistry.inbox(msg.queryId(),
msg.exchangeId());
if (inbox == null && msg.batchId() == 0) {
// first message sent before a fragment is built
// note that an inbox source fragment id is also used as an
exchange id
- Inbox<?> newInbox = new Inbox<>(baseInboxContext(nodeId,
msg.queryId(), msg.fragmentId()),
+ Inbox<?> newInbox = new Inbox<>(baseInboxContext(nodeName,
msg.queryId(), msg.fragmentId()),
this, mailboxRegistry, msg.exchangeId(), msg.exchangeId());
inbox = mailboxRegistry.register(newInbox);
@@ -200,22 +200,22 @@ public class ExchangeServiceImpl implements
ExchangeService {
if (inbox != null) {
try {
- inbox.onBatchReceived(nodeId, msg.batchId(), msg.last(),
Commons.cast(msg.rows()));
+ inbox.onBatchReceived(nodeName, msg.batchId(), msg.last(),
Commons.cast(msg.rows()));
} catch (Throwable e) {
inbox.onError(e);
throw new IgniteInternalException(UNEXPECTED_ERR, "Unexpected
exception", e);
}
} else if (LOG.isDebugEnabled()) {
- LOG.debug("Stale batch message received: [nodeId={}, queryId={},
fragmentId={}, exchangeId={}, batchId={}]",
- nodeId, msg.queryId(), msg.fragmentId(), msg.exchangeId(),
msg.batchId());
+ LOG.debug("Stale batch message received: [nodeName={}, queryId={},
fragmentId={}, exchangeId={}, batchId={}]",
+ nodeName, msg.queryId(), msg.fragmentId(),
msg.exchangeId(), msg.batchId());
}
}
/**
* Get minimal execution context to meet Inbox needs.
*/
- private ExecutionContext<?> baseInboxContext(String nodeId, UUID qryId,
long fragmentId) {
+ private ExecutionContext<?> baseInboxContext(String nodeName, UUID qryId,
long fragmentId) {
return new ExecutionContext<>(
BaseQueryContext.builder()
.logger(LOG)
@@ -223,7 +223,7 @@ public class ExchangeServiceImpl implements ExchangeService
{
taskExecutor,
qryId,
localNode,
- nodeId,
+ nodeName,
new FragmentDescription(
fragmentId,
null,
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionContext.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionContext.java
index 42bd2aa5f1..5d79cf740a 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionContext.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionContext.java
@@ -84,7 +84,7 @@ public class ExecutionContext<RowT> extends
AbstractQueryContext implements Data
private final ClusterNode localNode;
- private final String originatingNodeId;
+ private final String originatingNodeName;
private final RowHandler<RowT> handler;
@@ -106,13 +106,13 @@ public class ExecutionContext<RowT> extends
AbstractQueryContext implements Data
/**
* Constructor.
*
- * @param executor Task executor.
- * @param qctx Base query context.
- * @param qryId Query ID.
+ * @param executor Task executor.
+ * @param qctx Base query context.
+ * @param qryId Query ID.
* @param fragmentDesc Partitions information.
- * @param handler Row handler.
- * @param params Parameters.
- * @param tx Transaction.
+ * @param handler Row handler.
+ * @param params Parameters.
+ * @param tx Transaction.
*/
@SuppressWarnings("AssignmentOrReturnOfFieldWithMutableType")
public ExecutionContext(
@@ -120,7 +120,7 @@ public class ExecutionContext<RowT> extends
AbstractQueryContext implements Data
QueryTaskExecutor executor,
UUID qryId,
ClusterNode localNode,
- String originatingNodeId,
+ String originatingNodeName,
FragmentDescription fragmentDesc,
RowHandler<RowT> handler,
Map<String, Object> params,
@@ -135,7 +135,7 @@ public class ExecutionContext<RowT> extends
AbstractQueryContext implements Data
this.handler = handler;
this.params = params;
this.localNode = localNode;
- this.originatingNodeId = originatingNodeId;
+ this.originatingNodeName = originatingNodeName;
this.tx = tx;
expressionFactory = new ExpressionFactoryImpl<>(
@@ -220,10 +220,10 @@ public class ExecutionContext<RowT> extends
AbstractQueryContext implements Data
}
/**
- * Get originating node ID.
+ * Get originating node consistent ID.
*/
- public String originatingNodeId() {
- return originatingNodeId;
+ public String originatingNodeName() {
+ return originatingNodeName;
}
/**
@@ -290,7 +290,7 @@ public class ExecutionContext<RowT> extends
AbstractQueryContext implements Data
/**
* Sets correlated value.
*
- * @param id Correlation ID.
+ * @param id Correlation ID.
* @param value Correlated value.
*/
public void setCorrelated(@NotNull Object value, int id) {
@@ -323,8 +323,8 @@ public class ExecutionContext<RowT> extends
AbstractQueryContext implements Data
}
/**
- * Submits a Runnable task for execution and returns a Future representing
that task. The Future's {@code get} method will return {@code
- * null} upon <em>successful</em> completion.
+ * Submits a Runnable task for execution and returns a Future representing
that task. The Future's {@code get} method will return
+ * {@code null} upon <em>successful</em> completion.
*
* @param task the task to submit.
* @return a {@link CompletableFuture} representing pending task
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
index 81c70e51f4..540c9ec6bf 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImpl.java
@@ -289,8 +289,8 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
return new AsyncWrapper<>(res.iterator());
}
- private void onMessage(String nodeId, QueryStartRequest msg) {
- assert nodeId != null && msg != null;
+ private void onMessage(String nodeName, QueryStartRequest msg) {
+ assert nodeName != null && msg != null;
DistributedQueryManager queryManager =
queryManagerMap.computeIfAbsent(msg.queryId(), key -> {
BaseQueryContext ctx = createQueryContext(key, msg.schema(),
msg.parameters(), msg.txTime());
@@ -298,33 +298,33 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
return new DistributedQueryManager(ctx);
});
- queryManager.submitFragment(nodeId, msg.root(),
msg.fragmentDescription());
+ queryManager.submitFragment(nodeName, msg.root(),
msg.fragmentDescription());
}
- private void onMessage(String nodeId, QueryStartResponse msg) {
- assert nodeId != null && msg != null;
+ private void onMessage(String nodeName, QueryStartResponse msg) {
+ assert nodeName != null && msg != null;
DistributedQueryManager dqm = queryManagerMap.get(msg.queryId());
if (dqm != null) {
- dqm.acknowledgeFragment(nodeId, msg.fragmentId(), msg.error());
+ dqm.acknowledgeFragment(nodeName, msg.fragmentId(), msg.error());
}
}
- private void onMessage(String nodeId, ErrorMessage msg) {
- assert nodeId != null && msg != null;
+ private void onMessage(String nodeName, ErrorMessage msg) {
+ assert nodeName != null && msg != null;
DistributedQueryManager dqm = queryManagerMap.get(msg.queryId());
if (dqm != null) {
- RemoteException e = new RemoteException(nodeId, msg.queryId(),
msg.fragmentId(), msg.error());
+ RemoteException e = new RemoteException(nodeName, msg.queryId(),
msg.fragmentId(), msg.error());
dqm.onError(e);
}
}
- private void onMessage(String nodeId, QueryCloseMessage msg) {
- assert nodeId != null && msg != null;
+ private void onMessage(String nodeName, QueryCloseMessage msg) {
+ assert nodeName != null && msg != null;
DistributedQueryManager dqm = queryManagerMap.get(msg.queryId());
@@ -352,7 +352,7 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
/** {@inheritDoc} */
@Override
public void onDisappeared(ClusterNode member) {
- queryManagerMap.values().forEach(qm -> qm.onNodeLeft(member.id()));
+ queryManagerMap.values().forEach(qm -> qm.onNodeLeft(member.name()));
}
/** Returns local fragments for the query with given id. */
@@ -413,7 +413,9 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
return List.copyOf(localFragments);
}
- private void sendFragment(String targetNodeId, Fragment fragment,
FragmentDescription desc) throws IgniteInternalCheckedException {
+ private void sendFragment(
+ String targetNodeName, Fragment fragment, FragmentDescription
desc
+ ) throws IgniteInternalCheckedException {
QueryStartRequest req = FACTORY.queryStartRequest()
.queryId(ctx.queryId())
.fragmentId(fragment.fragmentId())
@@ -425,10 +427,10 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
.build();
var fut = new CompletableFuture<Void>();
- remoteFragmentInitCompletion.put(new
RemoteFragmentKey(targetNodeId, fragment.fragmentId()), fut);
+ remoteFragmentInitCompletion.put(new
RemoteFragmentKey(targetNodeName, fragment.fragmentId()), fut);
try {
- msgSrvc.send(targetNodeId, req);
+ msgSrvc.send(targetNodeName, req);
} catch (Exception ex) {
fut.complete(null);
@@ -440,7 +442,7 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
}
}
- private void acknowledgeFragment(String nodeId, long fragmentId,
@Nullable Throwable ex) {
+ private void acknowledgeFragment(String nodeName, long fragmentId,
@Nullable Throwable ex) {
if (ex != null) {
Long rootFragmentId0 = rootFragmentId;
@@ -455,7 +457,7 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
}
}
- remoteFragmentInitCompletion.get(new RemoteFragmentKey(nodeId,
fragmentId)).complete(null);
+ remoteFragmentInitCompletion.get(new RemoteFragmentKey(nodeName,
fragmentId)).complete(null);
}
private void onError(RemoteException ex) {
@@ -466,15 +468,16 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
});
}
- private void onNodeLeft(String nodeId) {
- remoteFragmentInitCompletion.entrySet().stream().filter(e ->
nodeId.equals(e.getKey().nodeId()))
+ private void onNodeLeft(String nodeName) {
+ remoteFragmentInitCompletion.entrySet().stream()
+ .filter(e -> nodeName.equals(e.getKey().nodeName()))
.forEach(e -> e.getValue()
.completeExceptionally(new IgniteInternalException(
- NODE_LEFT_ERR, "Node left the cluster
[nodeId=" + nodeId + "]")));
+ NODE_LEFT_ERR, "Node left the cluster
[nodeName=" + nodeName + "]")));
}
private void executeFragment(FragmentPlan plan, ExecutionContext<RowT>
ectx) {
- String origNodeId = ectx.originatingNodeId();
+ String origNodeName = ectx.originatingNodeName();
AbstractNode<RowT> node =
implementorFactory.create(ectx).go(plan.root());
@@ -503,14 +506,14 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
try {
msgSrvc.send(
- origNodeId,
+ origNodeName,
FACTORY.queryStartResponse()
.queryId(ectx.queryId())
.fragmentId(ectx.fragmentId())
.build()
);
} catch (IgniteInternalCheckedException e) {
- throw new IgniteInternalException(MESSAGE_SEND_ERR, "Failed to
send reply. [nodeId=" + origNodeId + ']', e);
+ throw new IgniteInternalException(MESSAGE_SEND_ERR, "Failed to
send reply. [nodeName=" + origNodeName + ']', e);
}
if (node instanceof Outbox) {
@@ -518,13 +521,13 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
}
}
- private ExecutionContext<RowT> createContext(String initiatorNodeId,
FragmentDescription desc) {
+ private ExecutionContext<RowT> createContext(String initiatorNodeName,
FragmentDescription desc) {
return new ExecutionContext<>(
ctx,
taskExecutor,
ctx.queryId(),
localNode,
- initiatorNodeId,
+ initiatorNodeName,
desc,
handler,
Commons.parametersMap(ctx.parameters()),
@@ -561,7 +564,7 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
private AsyncCursor<List<Object>> execute(MultiStepPlan plan) {
taskExecutor.execute(() -> {
- plan.init(mappingSrvc, new
MappingQueryContext(localNode.id()));
+ plan.init(mappingSrvc, new
MappingQueryContext(localNode.name()));
List<Fragment> fragments = plan.fragments();
@@ -585,8 +588,8 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
plan.remotes(fragment)
);
- for (String nodeId : fragmentDesc.nodeIds()) {
- sendFragment(nodeId, fragment, fragmentDesc);
+ for (String nodeName : fragmentDesc.nodeNames()) {
+ sendFragment(nodeName, fragment, fragmentDesc);
}
}
} catch (Throwable e) {
@@ -628,7 +631,7 @@ public class ExecutionServiceImpl<RowT> implements
ExecutionService, TopologyEve
.thenCompose(tmp -> {
Map<String, List<CompletableFuture<?>>>
requestsPerNode = new HashMap<>();
for (Map.Entry<RemoteFragmentKey,
CompletableFuture<Void>> entry : remoteFragmentInitCompletion.entrySet()) {
-
requestsPerNode.computeIfAbsent(entry.getKey().nodeId(), key -> new
ArrayList<>()).add(entry.getValue());
+
requestsPerNode.computeIfAbsent(entry.getKey().nodeName(), key -> new
ArrayList<>()).add(entry.getValue());
}
List<CompletableFuture<?>> cancelFuts = new
ArrayList<>();
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
index de2e94c06b..0c19d63b03 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/LogicalRelImplementor.java
@@ -190,9 +190,9 @@ public class LogicalRelImplementor<RowT> implements
IgniteRelVisitor<Node<RowT>>
IgniteDistribution distr = rel.distribution();
Destination<RowT> dest = distr.destination(ctx, affSrvc,
ctx.group(rel.sourceId()));
- String localNodeId = ctx.localNode().id();
+ String localNodeName = ctx.localNode().name();
- FilterNode<RowT> node = new FilterNode<>(ctx, r ->
Objects.equals(localNodeId, first(dest.targets(r))));
+ FilterNode<RowT> node = new FilterNode<>(ctx, r ->
Objects.equals(localNodeName, first(dest.targets(r))));
Node<RowT> input = visit(rel.getInput());
@@ -320,7 +320,7 @@ public class LogicalRelImplementor<RowT> implements
IgniteRelVisitor<Node<RowT>>
ColocationGroup group = ctx.group(rel.sourceId());
Comparator<RowT> comp = idx.type() == Type.SORTED ?
ctx.expressionFactory().comparator(outputCollation) : null;
- if (!group.nodeIds().contains(ctx.localNode().id())) {
+ if (!group.nodeNames().contains(ctx.localNode().name())) {
return new ScanNode<>(ctx, Collections.emptyList());
}
@@ -330,7 +330,7 @@ public class LogicalRelImplementor<RowT> implements
IgniteRelVisitor<Node<RowT>>
idx,
tbl,
rel.collation().getKeys(),
- group.partitions(ctx.localNode().id()),
+ group.partitions(ctx.localNode().name()),
comp,
ranges,
filters,
@@ -359,7 +359,7 @@ public class LogicalRelImplementor<RowT> implements
IgniteRelVisitor<Node<RowT>>
ColocationGroup group = ctx.group(rel.sourceId());
- if (!group.nodeIds().contains(ctx.localNode().id())) {
+ if (!group.nodeNames().contains(ctx.localNode().name())) {
return new ScanNode<>(ctx, Collections.emptyList());
}
@@ -367,7 +367,7 @@ public class LogicalRelImplementor<RowT> implements
IgniteRelVisitor<Node<RowT>>
ctx,
ctx.rowHandler().factory(ctx.getTypeFactory(), rowType),
tbl,
- group.partitions(ctx.localNode().id()),
+ group.partitions(ctx.localNode().name()),
filters,
prj,
requiredColumns == null ? null : requiredColumns.toBitSet()
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/MailboxRegistryImpl.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/MailboxRegistryImpl.java
index c004c45e78..2f00be6d6b 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/MailboxRegistryImpl.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/MailboxRegistryImpl.java
@@ -177,8 +177,8 @@ public class MailboxRegistryImpl implements
MailboxRegistry, TopologyEventHandle
/** {@inheritDoc} */
@Override
public void onDisappeared(ClusterNode member) {
- locals.values().forEach(n -> n.onNodeLeft(member.id()));
- remotes.values().forEach(n -> n.onNodeLeft(member.id()));
+ locals.values().forEach(n -> n.onNodeLeft(member.name()));
+ remotes.values().forEach(n -> n.onNodeLeft(member.name()));
}
private static class MailboxKey {
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/RemoteFragmentKey.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/RemoteFragmentKey.java
index 3ae6d665e0..312f3491c5 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/RemoteFragmentKey.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/RemoteFragmentKey.java
@@ -24,24 +24,24 @@ package org.apache.ignite.internal.sql.engine.exec;
* fragment id only to distinguish between remote fragments, the node id
should be attached.
*/
public class RemoteFragmentKey {
- private final String nodeId;
+ private final String nodeName;
private final long fragmentId;
/**
* Creates an object.
*
- * @param nodeId Id of the node that own a fragment.
+ * @param nodeName Consistent id of the node that own a fragment.
* @param fragmentId Id of the particular fragment owned by a remote node.
*/
- public RemoteFragmentKey(String nodeId, long fragmentId) {
- this.nodeId = nodeId;
+ public RemoteFragmentKey(String nodeName, long fragmentId) {
+ this.nodeName = nodeName;
this.fragmentId = fragmentId;
}
- /** Returns an id of the remote node. */
- public String nodeId() {
- return nodeId;
+ /** Returns the consistent id of a remote node. */
+ public String nodeName() {
+ return nodeName;
}
/** Returns an id os the fragment. */
@@ -63,12 +63,12 @@ public class RemoteFragmentKey {
if (fragmentId != that.fragmentId) {
return false;
}
- return nodeId.equals(that.nodeId);
+ return nodeName.equals(that.nodeName);
}
/** {@inheritDoc} */
@Override public int hashCode() {
- int res = nodeId.hashCode();
+ int res = nodeName.hashCode();
res = 31 * res + (int) (fragmentId ^ (fragmentId >>> 32));
return res;
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java
index 0069540164..c519c9a019 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Inbox.java
@@ -51,7 +51,7 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
private final Map<String, Buffer> perNodeBuffers;
- private volatile Collection<String> srcNodeIds;
+ private volatile Collection<String> srcNodeNames;
private Comparator<RowT> comp;
@@ -64,10 +64,10 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
/**
* Constructor.
*
- * @param ctx Execution context.
- * @param exchange Exchange service.
- * @param registry Mailbox registry.
- * @param exchangeId Exchange ID.
+ * @param ctx Execution context.
+ * @param exchange Exchange service.
+ * @param registry Mailbox registry.
+ * @param exchangeId Exchange ID.
* @param srcFragmentId Source fragment ID.
*/
public Inbox(
@@ -97,13 +97,13 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
* Inits this Inbox.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*
- * @param ctx Execution context.
- * @param srcNodeIds Source node IDs.
- * @param comp Optional comparator for merge exchange.
+ * @param ctx Execution context.
+ * @param srcNodeNames Source nodes' consistent IDs.
+ * @param comp Optional comparator for merge exchange.
*/
public void init(
- ExecutionContext<RowT> ctx, Collection<String> srcNodeIds,
@Nullable Comparator<RowT> comp) {
- assert srcNodeIds != null : "Collection srcNodeIds not found for
exchangeId: " + exchangeId;
+ ExecutionContext<RowT> ctx, Collection<String> srcNodeNames,
@Nullable Comparator<RowT> comp) {
+ assert srcNodeNames != null : "Collection srcNodeNames not found for
exchangeId: " + exchangeId;
assert context().fragmentId() == ctx.fragmentId() : "different
fragments unsupported: previous=" + context().fragmentId()
+ " current=" + ctx.fragmentId();
@@ -115,13 +115,13 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
this.comp = comp;
// memory barier
- this.srcNodeIds = new HashSet<>(srcNodeIds);
+ this.srcNodeNames = new HashSet<>(srcNodeNames);
}
/** {@inheritDoc} */
@Override
public void request(int rowsCnt) throws Exception {
- assert srcNodeIds != null;
+ assert srcNodeNames != null;
assert rowsCnt > 0 && requested == 0;
checkState();
@@ -162,13 +162,13 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
/**
* Pushes a batch into a buffer.
*
- * @param srcNodeId Source node id.
- * @param batchId Batch ID.
- * @param last Last batch flag.
- * @param rows Rows.
+ * @param srcNodeName Source node consistent id.
+ * @param batchId Batch ID.
+ * @param last Last batch flag.
+ * @param rows Rows.
*/
- public void onBatchReceived(String srcNodeId, int batchId, boolean last,
List<RowT> rows) throws Exception {
- Buffer buf = getOrCreateBuffer(srcNodeId);
+ public void onBatchReceived(String srcNodeName, int batchId, boolean last,
List<RowT> rows) throws Exception {
+ Buffer buf = getOrCreateBuffer(srcNodeName);
boolean waitingBefore = buf.check() == State.WAITING;
@@ -187,11 +187,11 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
private void push() throws Exception {
if (buffers == null) {
- for (String node : srcNodeIds) {
+ for (String node : srcNodeNames) {
checkNode(node);
}
- buffers = srcNodeIds.stream()
+ buffers = srcNodeNames.stream()
.map(this::getOrCreateBuffer)
.collect(Collectors.toList());
@@ -328,41 +328,41 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
}
}
- private void acknowledge(String nodeId, int batchId) throws
IgniteInternalCheckedException {
- exchange.acknowledge(nodeId, queryId(), srcFragmentId, exchangeId,
batchId);
+ private void acknowledge(String nodeName, int batchId) throws
IgniteInternalCheckedException {
+ exchange.acknowledge(nodeName, queryId(), srcFragmentId, exchangeId,
batchId);
}
- private Buffer getOrCreateBuffer(String nodeId) {
- return perNodeBuffers.computeIfAbsent(nodeId, this::createBuffer);
+ private Buffer getOrCreateBuffer(String nodeName) {
+ return perNodeBuffers.computeIfAbsent(nodeName, this::createBuffer);
}
- private Buffer createBuffer(String nodeId) {
- return new Buffer(nodeId);
+ private Buffer createBuffer(String nodeName) {
+ return new Buffer(nodeName);
}
/**
* OnNodeLeft.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- public void onNodeLeft(String nodeId) {
- if (context().originatingNodeId().equals(nodeId) && srcNodeIds ==
null) {
+ public void onNodeLeft(String nodeName) {
+ if (context().originatingNodeName().equals(nodeName) && srcNodeNames
== null) {
context().execute(this::close, this::onError);
- } else if (srcNodeIds != null && srcNodeIds.contains(nodeId)) {
- context().execute(() -> onNodeLeft0(nodeId), this::onError);
+ } else if (srcNodeNames != null && srcNodeNames.contains(nodeName)) {
+ context().execute(() -> onNodeLeft0(nodeName), this::onError);
}
}
- private void onNodeLeft0(String nodeId) throws Exception {
+ private void onNodeLeft0(String nodeName) throws Exception {
checkState();
- if (getOrCreateBuffer(nodeId).check() != State.END) {
- throw new IgniteInternalCheckedException(NODE_LEFT_ERR, "Failed to
execute query, node left [nodeId=" + nodeId + ']');
+ if (getOrCreateBuffer(nodeName).check() != State.END) {
+ throw new IgniteInternalCheckedException(NODE_LEFT_ERR, "Failed to
execute query, node left [nodeName=" + nodeName + ']');
}
}
- private void checkNode(String nodeId) throws
IgniteInternalCheckedException {
- if (!exchange.alive(nodeId)) {
- throw new IgniteInternalCheckedException(NODE_LEFT_ERR, "Failed to
execute query, node left [nodeId=" + nodeId + ']');
+ private void checkNode(String nodeName) throws
IgniteInternalCheckedException {
+ if (!exchange.alive(nodeName)) {
+ throw new IgniteInternalCheckedException(NODE_LEFT_ERR, "Failed to
execute query, node left [nodeName=" + nodeName + ']');
}
}
@@ -422,7 +422,7 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
private static final Batch<?> END = new Batch<>(0, false, null);
private final class Buffer {
- private final String nodeId;
+ private final String nodeName;
private int lastEnqueued = -1;
@@ -430,8 +430,8 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
private Batch<RowT> curr = waitingMark();
- private Buffer(String nodeId) {
- this.nodeId = nodeId;
+ private Buffer(String nodeName) {
+ this.nodeName = nodeName;
}
private void offer(int id, boolean last, List<RowT> rows) {
@@ -488,7 +488,7 @@ public class Inbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, Si
RowT row = curr.rows.set(curr.idx++, null);
if (curr.idx == curr.rows.size()) {
- acknowledge(nodeId, curr.batchId);
+ acknowledge(nodeName, curr.batchId);
if (!isEnd()) {
curr = pollBatch();
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Outbox.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Outbox.java
index 007463ac97..d18a2ef5c0 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Outbox.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/rel/Outbox.java
@@ -63,12 +63,12 @@ public class Outbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, S
* Constructor.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*
- * @param ctx Execution context.
- * @param exchange Exchange service.
- * @param registry Mailbox registry.
- * @param exchangeId Exchange ID.
+ * @param ctx Execution context.
+ * @param exchange Exchange service.
+ * @param registry Mailbox registry.
+ * @param exchangeId Exchange ID.
* @param targetFragmentId Target fragment ID.
- * @param dest Destination.
+ * @param dest Destination.
*/
public Outbox(
ExecutionContext<RowT> ctx,
@@ -95,15 +95,15 @@ public class Outbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, S
/**
* Callback method.
*
- * @param nodeId Target ID.
+ * @param nodeName Target consistent ID.
* @param batchId Batch ID.
*/
- public void onAcknowledge(String nodeId, int batchId) throws Exception {
- assert nodeBuffers.containsKey(nodeId);
+ public void onAcknowledge(String nodeName, int batchId) throws Exception {
+ assert nodeBuffers.containsKey(nodeName);
checkState();
- nodeBuffers.get(nodeId).acknowledge(batchId);
+ nodeBuffers.get(nodeName).acknowledge(batchId);
}
/**
@@ -153,7 +153,8 @@ public class Outbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, S
}
/** {@inheritDoc} */
- @Override public void onError(Throwable e) {
+ @Override
+ public void onError(Throwable e) {
try {
sendError(e);
} catch (IgniteInternalCheckedException ex) {
@@ -198,28 +199,28 @@ public class Outbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, S
return this;
}
- private void sendBatch(String nodeId, int batchId, boolean last,
List<RowT> rows) throws IgniteInternalCheckedException {
- exchange.sendBatch(nodeId, queryId(), targetFragmentId, exchangeId,
batchId, last, rows);
+ private void sendBatch(String nodeName, int batchId, boolean last,
List<RowT> rows) throws IgniteInternalCheckedException {
+ exchange.sendBatch(nodeName, queryId(), targetFragmentId, exchangeId,
batchId, last, rows);
}
private void sendError(Throwable err) throws
IgniteInternalCheckedException {
- exchange.sendError(context().originatingNodeId(), queryId(),
fragmentId(), err);
+ exchange.sendError(context().originatingNodeName(), queryId(),
fragmentId(), err);
}
- private void sendInboxClose(String nodeId) {
+ private void sendInboxClose(String nodeName) {
try {
- exchange.closeInbox(nodeId, queryId(), targetFragmentId,
exchangeId);
+ exchange.closeInbox(nodeName, queryId(), targetFragmentId,
exchangeId);
} catch (IgniteInternalCheckedException e) {
LOG.info("Unable to send cancel message", e);
}
}
- private Buffer getOrCreateBuffer(String nodeId) {
- return nodeBuffers.computeIfAbsent(nodeId, this::createBuffer);
+ private Buffer getOrCreateBuffer(String nodeName) {
+ return nodeBuffers.computeIfAbsent(nodeName, this::createBuffer);
}
- private Buffer createBuffer(String nodeId) {
- return new Buffer(nodeId);
+ private Buffer createBuffer(String nodeName) {
+ return new Buffer(nodeName);
}
private void flush() throws Exception {
@@ -258,14 +259,14 @@ public class Outbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, S
* OnNodeLeft.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- public void onNodeLeft(String nodeId) {
- if (nodeId.equals(context().originatingNodeId())) {
+ public void onNodeLeft(String nodeName) {
+ if (nodeName.equals(context().originatingNodeName())) {
context().execute(this::close, this::onError);
}
}
private final class Buffer {
- private final String nodeId;
+ private final String nodeName;
private int hwm = -1;
@@ -273,8 +274,8 @@ public class Outbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, S
private List<RowT> curr;
- private Buffer(String nodeId) {
- this.nodeId = nodeId;
+ private Buffer(String nodeName) {
+ this.nodeName = nodeName;
curr = new ArrayList<>(IO_BATCH_SIZE);
}
@@ -301,7 +302,7 @@ public class Outbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, S
assert ready();
if (curr.size() == IO_BATCH_SIZE) {
- sendBatch(nodeId, ++hwm, false, curr);
+ sendBatch(nodeName, ++hwm, false, curr);
curr = new ArrayList<>(IO_BATCH_SIZE);
}
@@ -323,7 +324,7 @@ public class Outbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, S
List<RowT> tmp = curr;
curr = null;
- sendBatch(nodeId, batchId, true, tmp);
+ sendBatch(nodeName, batchId, true, tmp);
}
/**
@@ -357,7 +358,7 @@ public class Outbox<RowT> extends AbstractNode<RowT>
implements Mailbox<RowT>, S
curr = null;
if (currBatchId >= 0) {
- sendInboxClose(nodeId);
+ sendInboxClose(nodeName);
}
}
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageListener.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageListener.java
index f45dce72e0..555d2687e2 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageListener.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageListener.java
@@ -27,8 +27,8 @@ public interface MessageListener {
/**
* OnMessage.
*
- * @param nodeId Sender node ID.
- * @param msg Message.
+ * @param nodeName Sender node consistent ID.
+ * @param msg Message.
*/
- void onMessage(String nodeId, NetworkMessage msg);
+ void onMessage(String nodeName, NetworkMessage msg);
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageService.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageService.java
index c4ba4fe88c..633ef7f86d 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageService.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageService.java
@@ -25,28 +25,27 @@ import org.apache.ignite.network.NetworkMessage;
* MessageService interface.
*/
// TODO: Documentation https://issues.apache.org/jira/browse/IGNITE-15859
-// TODO: Service should operate with consistent IDs, see
https://issues.apache.org/jira/browse/IGNITE-18063
public interface MessageService extends LifecycleAware {
/**
* Sends a message to given node.
*
- * @param nodeId Node ID.
- * @param msg Message.
+ * @param nodeName Node consistent ID.
+ * @param msg Message.
*/
- void send(String nodeId, NetworkMessage msg) throws
IgniteInternalCheckedException;
+ void send(String nodeName, NetworkMessage msg) throws
IgniteInternalCheckedException;
/**
* Checks whether a node with given ID is alive.
*
- * @param nodeId Node ID.
+ * @param nodeName Node consistent ID.
* @return {@code True} if node is alive.
*/
- boolean alive(String nodeId);
+ boolean alive(String nodeName);
/**
* Registers a listener for messages of a given type.
*
- * @param lsnr Listener.
+ * @param lsnr Listener.
* @param msgId Message id.
*/
void register(MessageListener lsnr, short msgId);
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageServiceImpl.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageServiceImpl.java
index d74516862f..425a5c5c68 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageServiceImpl.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/message/MessageServiceImpl.java
@@ -43,7 +43,7 @@ public class MessageServiceImpl implements MessageService {
private final MessagingService messagingSrvc;
- private final String locNodeId;
+ private final String locNodeName;
private final QueryTaskExecutor taskExecutor;
@@ -66,7 +66,7 @@ public class MessageServiceImpl implements MessageService {
this.taskExecutor = taskExecutor;
this.busyLock = busyLock;
- locNodeId = topSrvc.localMember().id();
+ locNodeName = topSrvc.localMember().name();
}
/** {@inheritDoc} */
@@ -77,20 +77,22 @@ public class MessageServiceImpl implements MessageService {
/** {@inheritDoc} */
@Override
- public void send(String nodeId, NetworkMessage msg) throws
IgniteInternalCheckedException {
+ public void send(String nodeName, NetworkMessage msg) throws
IgniteInternalCheckedException {
if (!busyLock.enterBusy()) {
return;
}
try {
- if (locNodeId.equals(nodeId)) {
- onMessage(nodeId, msg);
+ if (locNodeName.equals(nodeName)) {
+ onMessage(nodeName, msg);
} else {
- ClusterNode node = topSrvc.allMembers().stream()
- .filter(cn -> nodeId.equals(cn.id()))
- .findFirst()
- .orElseThrow(() -> new IgniteInternalException(
- NODE_LEFT_ERR, "Failed to send message to node
(has node left grid?): " + nodeId));
+ ClusterNode node = topSrvc.getByConsistentId(nodeName);
+
+ if (node == null) {
+ throw new IgniteInternalException(
+ NODE_LEFT_ERR, "Failed to send message to node
(has node left grid?): " + nodeName
+ );
+ }
try {
messagingSrvc.send(node, msg).join();
@@ -121,18 +123,16 @@ public class MessageServiceImpl implements MessageService
{
/** {@inheritDoc} */
@Override
- public boolean alive(String nodeId) {
- return topSrvc.allMembers().stream()
- .map(ClusterNode::id)
- .anyMatch(id -> id.equals(nodeId));
+ public boolean alive(String nodeName) {
+ return topSrvc.getByConsistentId(nodeName) != null;
}
- private void onMessage(String nodeId, NetworkMessage msg) {
+ private void onMessage(String consistentId, NetworkMessage msg) {
if (msg instanceof ExecutionContextAwareMessage) {
ExecutionContextAwareMessage msg0 = (ExecutionContextAwareMessage)
msg;
- taskExecutor.execute(msg0.queryId(), msg0.fragmentId(), () ->
onMessageInternal(nodeId, msg));
+ taskExecutor.execute(msg0.queryId(), msg0.fragmentId(), () ->
onMessageInternal(consistentId, msg));
} else {
- taskExecutor.execute(() -> onMessageInternal(nodeId, msg));
+ taskExecutor.execute(() -> onMessageInternal(consistentId, msg));
}
}
@@ -144,13 +144,13 @@ public class MessageServiceImpl implements MessageService
{
try {
assert msg.groupType() == GROUP_TYPE : "unexpected message group
grpType=" + msg.groupType();
- onMessage(sender.id(), msg);
+ onMessage(sender.name(), msg);
} finally {
busyLock.leaveBusy();
}
}
- private void onMessageInternal(String nodeId, NetworkMessage msg) {
+ private void onMessageInternal(String consistentId, NetworkMessage msg) {
if (!busyLock.enterBusy()) {
return;
}
@@ -161,7 +161,7 @@ public class MessageServiceImpl implements MessageService {
"there is no listener for msgType=" + msg.messageType()
);
- lsnr.onMessage(nodeId, msg);
+ lsnr.onMessage(consistentId, msg);
} finally {
busyLock.leaveBusy();
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/ColocationGroup.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/ColocationGroup.java
index 8a08a4bf2a..21f8f08bfb 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/ColocationGroup.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/ColocationGroup.java
@@ -42,18 +42,18 @@ public class ColocationGroup implements Serializable {
private static final int SYNTHETIC_PARTITIONS_COUNT = 512;
// TODO:
IgniteSystemProperties.getInteger("IGNITE_CALCITE_SYNTHETIC_PARTITIONS_COUNT",
512);
- private List<Long> sourceIds;
+ private final List<Long> sourceIds;
- private List<String> nodeIds;
+ private final List<String> nodeNames;
- private List<List<String>> assignments;
+ private final List<List<String>> assignments;
/**
* ForNodes.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- public static ColocationGroup forNodes(List<String> nodeIds) {
- return new ColocationGroup(null, nodeIds, null);
+ public static ColocationGroup forNodes(List<String> nodeNames) {
+ return new ColocationGroup(null, nodeNames, null);
}
/**
@@ -76,9 +76,9 @@ public class ColocationGroup implements Serializable {
* Constructor.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- private ColocationGroup(List<Long> sourceIds, List<String> nodeIds,
List<List<String>> assignments) {
+ private ColocationGroup(List<Long> sourceIds, List<String> nodeNames,
List<List<String>> assignments) {
this.sourceIds = sourceIds;
- this.nodeIds = nodeIds;
+ this.nodeNames = nodeNames;
this.assignments = assignments;
}
@@ -92,8 +92,8 @@ public class ColocationGroup implements Serializable {
/**
* Get lists of nodes capable to execute a query fragment for what the
mapping is calculated.
*/
- public List<String> nodeIds() {
- return nodeIds == null ? Collections.emptyList() : nodeIds;
+ public List<String> nodeNames() {
+ return nodeNames == null ? Collections.emptyList() : nodeNames;
}
/**
@@ -139,14 +139,14 @@ public class ColocationGroup implements Serializable {
sourceIds = Commons.combine(this.sourceIds, other.sourceIds);
}
- List<String> nodeIds;
- if (this.nodeIds == null || other.nodeIds == null) {
- nodeIds = firstNotNull(this.nodeIds, other.nodeIds);
+ List<String> nodeNames;
+ if (this.nodeNames == null || other.nodeNames == null) {
+ nodeNames = firstNotNull(this.nodeNames, other.nodeNames);
} else {
- nodeIds = Commons.intersect(other.nodeIds, this.nodeIds);
+ nodeNames = Commons.intersect(other.nodeNames, this.nodeNames);
}
- if (nodeIds != null && nodeIds.isEmpty()) {
+ if (nodeNames != null && nodeNames.isEmpty()) {
throw new ColocationMappingException("Failed to map fragment to
location. "
+ "Replicated query parts are not co-located on all
nodes");
}
@@ -155,8 +155,8 @@ public class ColocationGroup implements Serializable {
if (this.assignments == null || other.assignments == null) {
assignments = firstNotNull(this.assignments, other.assignments);
- if (assignments != null && nodeIds != null) {
- Set<String> filter = new HashSet<>(nodeIds);
+ if (assignments != null && nodeNames != null) {
+ Set<String> filter = new HashSet<>(nodeNames);
List<List<String>> assignments0 = new
ArrayList<>(assignments.size());
for (int i = 0; i < assignments.size(); i++) {
@@ -175,7 +175,7 @@ public class ColocationGroup implements Serializable {
} else {
assert this.assignments.size() == other.assignments.size();
assignments = new ArrayList<>(this.assignments.size());
- Set<String> filter = nodeIds == null ? null : new
HashSet<>(nodeIds);
+ Set<String> filter = nodeNames == null ? null : new
HashSet<>(nodeNames);
for (int i = 0; i < this.assignments.size(); i++) {
List<String> assignment =
Commons.intersect(this.assignments.get(i), other.assignments.get(i));
@@ -191,7 +191,7 @@ public class ColocationGroup implements Serializable {
}
}
- return new ColocationGroup(sourceIds, nodeIds, assignments);
+ return new ColocationGroup(sourceIds, nodeNames, assignments);
}
/**
@@ -199,7 +199,7 @@ public class ColocationGroup implements Serializable {
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
public ColocationGroup finalaze() {
- if (assignments == null && nodeIds == null) {
+ if (assignments == null && nodeNames == null) {
return this;
}
@@ -217,38 +217,38 @@ public class ColocationGroup implements Serializable {
return new ColocationGroup(sourceIds, new ArrayList<>(nodes),
assignments);
}
- return forNodes0(nodeIds);
+ return forNodes0(nodeNames);
}
/**
* MapToNodes.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- public ColocationGroup mapToNodes(List<String> nodeIds) {
- return !nullOrEmpty(this.nodeIds) ? this : forNodes0(nodeIds);
+ public ColocationGroup mapToNodes(List<String> nodeNames) {
+ return !nullOrEmpty(this.nodeNames) ? this : forNodes0(nodeNames);
}
@NotNull
- private ColocationGroup forNodes0(List<String> nodeIds) {
+ private ColocationGroup forNodes0(List<String> nodeNames) {
List<List<String>> assignments = new
ArrayList<>(SYNTHETIC_PARTITIONS_COUNT);
for (int i = 0; i < SYNTHETIC_PARTITIONS_COUNT; i++) {
- assignments.add(asList(nodeIds.get(i % nodeIds.size())));
+ assignments.add(asList(nodeNames.get(i % nodeNames.size())));
}
- return new ColocationGroup(sourceIds, nodeIds, assignments);
+ return new ColocationGroup(sourceIds, nodeNames, assignments);
}
/**
* Returns List of partitions to scan on the given node.
*
- * @param nodeId Cluster node ID.
+ * @param nodeNames Cluster node consistent ID.
* @return List of partitions to scan on the given node.
*/
- public int[] partitions(String nodeId) {
+ public int[] partitions(String nodeNames) {
IgniteIntList parts = new IgniteIntList(assignments.size());
for (int i = 0; i < assignments.size(); i++) {
List<String> assignment = assignments.get(i);
- if (Objects.equals(nodeId, first(assignment))) {
+ if (Objects.equals(nodeNames, first(assignment))) {
parts.add(i);
}
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/FragmentDescription.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/FragmentDescription.java
index abd28b6965..a17c0c64f9 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/FragmentDescription.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/FragmentDescription.java
@@ -63,8 +63,8 @@ public class FragmentDescription implements Serializable {
/**
* Get node ids.
*/
- public List<String> nodeIds() {
- return mapping.nodeIds();
+ public List<String> nodeNames() {
+ return mapping.nodeNames();
}
/**
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/FragmentMapping.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/FragmentMapping.java
index e1164da0b4..6951bb0fe7 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/FragmentMapping.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/FragmentMapping.java
@@ -64,8 +64,8 @@ public class FragmentMapping implements Serializable {
* Create.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- public static FragmentMapping create(String nodeId) {
- return new
FragmentMapping(ColocationGroup.forNodes(Collections.singletonList(nodeId)));
+ public static FragmentMapping create(String nodeName) {
+ return new
FragmentMapping(ColocationGroup.forNodes(Collections.singletonList(nodeName)));
}
/**
@@ -139,9 +139,9 @@ public class FragmentMapping implements Serializable {
* NodeIds.
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
- public List<String> nodeIds() {
+ public List<String> nodeNames() {
return colocationGroups.stream()
- .flatMap(g -> g.nodeIds().stream())
+ .flatMap(g -> g.nodeNames().stream())
.distinct().collect(Collectors.toList());
}
@@ -158,7 +158,7 @@ public class FragmentMapping implements Serializable {
colocationGroups = Commons.transform(colocationGroups,
ColocationGroup::finalaze);
- List<String> nodes = nodeIds();
+ List<String> nodes = nodeNames();
List<String> nodes0 = nodes.isEmpty() ? nodesSource.get() : nodes;
colocationGroups = Commons.transform(colocationGroups, g ->
g.mapToNodes(nodes0));
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdFragmentMapping.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdFragmentMapping.java
index b68eeccf04..33d9bf189d 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdFragmentMapping.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/IgniteMdFragmentMapping.java
@@ -237,7 +237,7 @@ public class IgniteMdFragmentMapping implements
MetadataHandler<FragmentMappingM
List<List<String>> fakeAssignments = new
ArrayList<>(group.assignments().size());
for (int i = 0; i < group.assignments().size(); i++) {
- fakeAssignments.add(List.of(ctx.localNodeId()));
+ fakeAssignments.add(List.of(ctx.locNodeName()));
}
group = ColocationGroup.forAssignments(fakeAssignments);
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/MappingServiceImpl.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/MappingServiceImpl.java
index 11d600ca12..c0e81d2803 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/MappingServiceImpl.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/MappingServiceImpl.java
@@ -58,6 +58,6 @@ public class MappingServiceImpl implements MappingService {
throw new IllegalStateException("failed to map query to execution
nodes. Nodes list is empty.");
}
- return Commons.transform(nodes, ClusterNode::id);
+ return Commons.transform(nodes, ClusterNode::name);
}
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/RemoteException.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/RemoteException.java
index 4139b8edcc..95822b032c 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/RemoteException.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/metadata/RemoteException.java
@@ -24,7 +24,7 @@ import java.util.UUID;
* TODO Documentation https://issues.apache.org/jira/browse/IGNITE-15859
*/
public class RemoteException extends RuntimeException {
- private final String nodeId;
+ private final String nodeName;
private final UUID queryId;
@@ -33,23 +33,23 @@ public class RemoteException extends RuntimeException {
/**
* Constructor.
*
- * @param cause Cause.
- * @param nodeId Node ID.
- * @param queryId Query ID.
+ * @param cause Cause.
+ * @param nodeName Node consistent ID.
+ * @param queryId Query ID.
* @param fragmentId Fragment ID.
*/
- public RemoteException(String nodeId, UUID queryId, long fragmentId,
Throwable cause) {
+ public RemoteException(String nodeName, UUID queryId, long fragmentId,
Throwable cause) {
super("Remote query execution", cause);
- this.nodeId = nodeId;
+ this.nodeName = nodeName;
this.queryId = queryId;
this.fragmentId = fragmentId;
}
/**
- * Get node ID.
+ * Get node consistent ID.
*/
- public String nodeId() {
- return nodeId;
+ public String nodeName() {
+ return nodeName;
}
/**
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/AbstractMultiStepPlan.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/AbstractMultiStepPlan.java
index a787aadbc7..d8268617c0 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/AbstractMultiStepPlan.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/AbstractMultiStepPlan.java
@@ -96,7 +96,7 @@ public abstract class AbstractMultiStepPlan implements
MultiStepPlan {
Long2ObjectOpenHashMap<List<String>> res = new
Long2ObjectOpenHashMap<>(capacity(remotes.size()));
for (IgniteReceiver remote : remotes) {
- res.put(remote.exchangeId(),
mapping(remote.sourceFragmentId()).nodeIds());
+ res.put(remote.exchangeId(),
mapping(remote.sourceFragmentId()).nodeNames());
}
return res;
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/Fragment.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/Fragment.java
index bcce4f89ac..81daa573c4 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/Fragment.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/Fragment.java
@@ -111,15 +111,15 @@ public class Fragment {
FragmentMapping mapping =
IgniteMdFragmentMapping.fragmentMappingForMetadataQuery(root, mq, ctx);
if (rootFragment()) {
- mapping =
FragmentMapping.create(ctx.localNodeId()).colocate(mapping);
+ mapping =
FragmentMapping.create(ctx.locNodeName()).colocate(mapping);
}
- if (single() && mapping.nodeIds().size() > 1) {
+ if (single() && mapping.nodeNames().size() > 1) {
// this is possible when the fragment contains scan of a
replicated cache, which brings
// several nodes (actually all containing nodes) to the
colocation group, but this fragment
// supposed to be executed on a single node, so let's choose
one wisely
- mapping = FragmentMapping.create(mapping.nodeIds()
-
.get(ThreadLocalRandom.current().nextInt(mapping.nodeIds().size()))).colocate(mapping);
+ mapping = FragmentMapping.create(mapping.nodeNames()
+
.get(ThreadLocalRandom.current().nextInt(mapping.nodeNames().size()))).colocate(mapping);
}
return mapping.finalize(nodesSource);
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/MappingQueryContext.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/MappingQueryContext.java
index d4cd8a8e44..74204d236f 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/MappingQueryContext.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/prepare/MappingQueryContext.java
@@ -27,17 +27,17 @@ import org.apache.ignite.internal.sql.engine.util.Commons;
* Query mapping context.
*/
public class MappingQueryContext {
- private final String locNodeId;
+ private final String locNodeName;
private RelOptCluster cluster;
/**
* Constructor.
*
- * @param locNodeId Local node identifier.
+ * @param locNodeName Local node consistent ID.
*/
- public MappingQueryContext(String locNodeId) {
- this.locNodeId = locNodeId;
+ public MappingQueryContext(String locNodeName) {
+ this.locNodeName = locNodeName;
}
/** Creates a cluster. */
@@ -52,7 +52,7 @@ public class MappingQueryContext {
return cluster;
}
- public String localNodeId() {
- return locNodeId;
+ public String locNodeName() {
+ return locNodeName;
}
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/trait/DistributionFunction.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/trait/DistributionFunction.java
index 3df332c31f..697a54502e 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/trait/DistributionFunction.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/trait/DistributionFunction.java
@@ -174,9 +174,9 @@ public abstract class DistributionFunction {
@Override
public <RowT> Destination<RowT> destination(ExecutionContext<RowT>
ctx, AffinityService affinityService,
ColocationGroup m, ImmutableIntList k) {
- assert m != null && !nullOrEmpty(m.nodeIds());
+ assert m != null && !nullOrEmpty(m.nodeNames());
- return new AllNodes<>(m.nodeIds());
+ return new AllNodes<>(m.nodeNames());
}
}
@@ -193,9 +193,9 @@ public abstract class DistributionFunction {
@Override
public <RowT> Destination<RowT> destination(ExecutionContext<RowT>
ctx, AffinityService affinityService,
ColocationGroup m, ImmutableIntList k) {
- assert m != null && !nullOrEmpty(m.nodeIds());
+ assert m != null && !nullOrEmpty(m.nodeNames());
- return new RandomNode<>(m.nodeIds());
+ return new RandomNode<>(m.nodeNames());
}
}
@@ -212,11 +212,11 @@ public abstract class DistributionFunction {
@Override
public <RowT> Destination<RowT> destination(ExecutionContext<RowT>
ctx, AffinityService affinityService,
ColocationGroup m, ImmutableIntList k) {
- if (m == null || m.nodeIds() == null || m.nodeIds().size() != 1) {
+ if (m == null || m.nodeNames() == null || m.nodeNames().size() !=
1) {
throw new IllegalStateException();
}
- return new
AllNodes<>(Collections.singletonList(Objects.requireNonNull(first(m.nodeIds()))));
+ return new
AllNodes<>(Collections.singletonList(Objects.requireNonNull(first(m.nodeNames()))));
}
}
diff --git
a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
index 7c95dad225..3cec811cb1 100644
---
a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
+++
b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/StopCalciteModuleTest.java
@@ -36,7 +36,6 @@ import java.lang.management.ManagementFactory;
import java.lang.management.ThreadInfo;
import java.lang.management.ThreadMXBean;
import java.util.Arrays;
-import java.util.Collections;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
@@ -45,8 +44,6 @@ import java.util.concurrent.Flow;
import java.util.function.Consumer;
import java.util.function.Function;
import org.apache.ignite.configuration.ConfigurationValue;
-import
org.apache.ignite.internal.configuration.testframework.ConfigurationExtension;
-import
org.apache.ignite.internal.configuration.testframework.InjectConfiguration;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.index.IndexManager;
import org.apache.ignite.internal.logger.IgniteLogger;
@@ -59,7 +56,6 @@ import org.apache.ignite.internal.schema.SchemaDescriptor;
import org.apache.ignite.internal.schema.SchemaManager;
import org.apache.ignite.internal.schema.SchemaRegistry;
import org.apache.ignite.internal.schema.configuration.TableConfiguration;
-import org.apache.ignite.internal.schema.configuration.TablesConfiguration;
import org.apache.ignite.internal.schema.registry.SchemaRegistryImpl;
import org.apache.ignite.internal.schema.row.RowAssembler;
import org.apache.ignite.internal.sql.engine.exec.ExecutionCancelledException;
@@ -91,7 +87,6 @@ import org.mockito.junit.jupiter.MockitoExtension;
/**
* Stop Calcite module test.
*/
-@ExtendWith(ConfigurationExtension.class)
@ExtendWith(MockitoExtension.class)
public class StopCalciteModuleTest {
/** The logger. */
@@ -102,41 +97,38 @@ public class StopCalciteModuleTest {
private static final String NODE_NAME = "mock-node-name";
@Mock
- ClusterService clusterSrvc;
+ private ClusterService clusterSrvc;
@Mock
- TableManager tableManager;
+ private TableManager tableManager;
@Mock
- IndexManager indexManager;
+ private IndexManager indexManager;
@Mock
- SchemaManager schemaManager;
+ private SchemaManager schemaManager;
@Mock
- DataStorageManager dataStorageManager;
+ private DataStorageManager dataStorageManager;
@Mock
- MessagingService msgSrvc;
+ private MessagingService msgSrvc;
@Mock
- TxManager txManager;
+ private TxManager txManager;
@Mock
- TopologyService topologySrvc;
+ private TopologyService topologySrvc;
@Mock
- InternalTable tbl;
+ private InternalTable tbl;
@Mock
- HybridClock clock;
+ private HybridClock clock;
- SchemaRegistry schemaReg;
+ private SchemaRegistry schemaReg;
- TestRevisionRegister testRevisionRegister = new TestRevisionRegister();
-
- @InjectConfiguration
- TablesConfiguration tablesConfig;
+ private final TestRevisionRegister testRevisionRegister = new
TestRevisionRegister();
/**
* Before.
@@ -149,7 +141,6 @@ public class StopCalciteModuleTest {
ClusterNode node = new ClusterNode("mock-node-id", NODE_NAME, null);
when(topologySrvc.localMember()).thenReturn(node);
-
when(topologySrvc.allMembers()).thenReturn(Collections.singleton(node));
SchemaDescriptor schemaDesc = new SchemaDescriptor(
1,
diff --git
a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
index e7a90739a4..4931e073de 100644
---
a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
+++
b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/ExecutionServiceImplTest.java
@@ -39,6 +39,7 @@ import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Queue;
+import java.util.UUID;
import java.util.concurrent.CompletionException;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
@@ -97,12 +98,12 @@ public class ExecutionServiceImplTest {
/** Timeout in ms for async operations. */
private static final long TIMEOUT_IN_MS = 2_000;
- private final List<String> nodeIds = List.of("node_1", "node_2", "node_3");
+ private final List<String> nodeNames = List.of("node_1", "node_2",
"node_3");
private final Map<String, List<Object[]>> dataPerNode = Map.of(
- nodeIds.get(0), List.of(new Object[]{0, 0}, new Object[]{3, 3},
new Object[]{6, 6}),
- nodeIds.get(1), List.of(new Object[]{1, 1}, new Object[]{4, 4},
new Object[]{7, 7}),
- nodeIds.get(2), List.of(new Object[]{2, 2}, new Object[]{5, 5},
new Object[]{8, 8})
+ nodeNames.get(0), List.of(new Object[]{0, 0}, new Object[]{3, 3},
new Object[]{6, 6}),
+ nodeNames.get(1), List.of(new Object[]{1, 1}, new Object[]{4, 4},
new Object[]{7, 7}),
+ nodeNames.get(2), List.of(new Object[]{2, 2}, new Object[]{5, 5},
new Object[]{8, 8})
);
private final TestTable table = createTable("TEST_TBL", 1_000_000,
IgniteDistributions.random(),
@@ -117,7 +118,7 @@ public class ExecutionServiceImplTest {
@BeforeEach
public void init() {
testCluster = new TestCluster();
- executionServices =
nodeIds.stream().map(this::create).collect(Collectors.toList());
+ executionServices =
nodeNames.stream().map(this::create).collect(Collectors.toList());
prepareService = new PrepareServiceImpl("test", 0, null);
prepareService.start();
@@ -137,7 +138,7 @@ public class ExecutionServiceImplTest {
var ctx = createContext();
var plan = prepare("SELECT * FROM test_tbl", ctx);
- nodeIds.stream().map(testCluster::node).forEach(TestNode::pauseScan);
+ nodeNames.stream().map(testCluster::node).forEach(TestNode::pauseScan);
var cursor = execService.executePlan(plan, ctx);
@@ -171,7 +172,7 @@ public class ExecutionServiceImplTest {
var ctx = createContext();
var plan = prepare("SELECT * FROM test_tbl", ctx);
- nodeIds.stream().map(testCluster::node).forEach(TestNode::pauseScan);
+ nodeNames.stream().map(testCluster::node).forEach(TestNode::pauseScan);
var cursor = execService.executePlan(plan, ctx);
@@ -205,14 +206,14 @@ public class ExecutionServiceImplTest {
var ctx = createContext();
var plan = prepare("SELECT * FROM test_tbl", ctx);
- nodeIds.stream().map(testCluster::node).forEach(TestNode::pauseScan);
+ nodeNames.stream().map(testCluster::node).forEach(TestNode::pauseScan);
var expectedEx = new RuntimeException("Test error");
- testCluster.node(nodeIds.get(2)).interceptor((nodeId, msg, original)
-> {
+ testCluster.node(nodeNames.get(2)).interceptor((nodeName, msg,
original) -> {
if (msg instanceof QueryStartRequest) {
try {
-
testCluster.node(nodeIds.get(2)).messageService().send(nodeId, new
SqlQueryMessagesFactory().queryStartResponse()
+
testCluster.node(nodeNames.get(2)).messageService().send(nodeName, new
SqlQueryMessagesFactory().queryStartResponse()
.queryId(((QueryStartRequest) msg).queryId())
.fragmentId(((QueryStartRequest) msg).fragmentId())
.error(expectedEx)
@@ -222,7 +223,7 @@ public class ExecutionServiceImplTest {
throw new
IgniteInternalException(OPERATION_INTERRUPTED_ERR, e);
}
} else {
- original.onMessage(nodeId, msg);
+ original.onMessage(nodeName, msg);
}
});
@@ -254,7 +255,7 @@ public class ExecutionServiceImplTest {
var ctx = createContext();
var plan = prepare("SELECT * FROM test_tbl", ctx);
- nodeIds.stream().map(testCluster::node).forEach(TestNode::pauseScan);
+ nodeNames.stream().map(testCluster::node).forEach(TestNode::pauseScan);
var cursor = execService.executePlan(plan, ctx);
@@ -327,36 +328,32 @@ public class ExecutionServiceImplTest {
.mapToInt(i -> i).sum() == 0, TIMEOUT_IN_MS));
}
- /** Creates an execution service instance for the node with given id. */
- public ExecutionServiceImpl<Object[]> create(String nodeId) {
- if (!nodeIds.contains(nodeId)) {
- throw new IllegalArgumentException(format("Node id should be one
of {}, but was '{}'", nodeIds, nodeId));
+ /** Creates an execution service instance for the node with given
consistent id. */
+ public ExecutionServiceImpl<Object[]> create(String nodeName) {
+ if (!nodeNames.contains(nodeName)) {
+ throw new IllegalArgumentException(format("Node id should be one
of {}, but was '{}'", nodeNames, nodeName));
}
- var taskExecutor = new QueryTaskExecutorImpl(nodeId);
+ var taskExecutor = new QueryTaskExecutorImpl(nodeName);
- var node = testCluster.addNode(nodeId, taskExecutor);
+ var node = testCluster.addNode(nodeName, taskExecutor);
- node.dataset(dataPerNode.get(nodeId));
+ node.dataset(dataPerNode.get(nodeName));
var messageService = node.messageService();
var mailboxRegistry = new MailboxRegistryImpl();
+ var clusterNode = new ClusterNode(UUID.randomUUID().toString(),
nodeName, NetworkAddress.from("127.0.0.1:1111"));
- var exchangeService = new ExchangeServiceImpl(
- new ClusterNode(nodeId, "fake-test-node",
NetworkAddress.from("127.0.0.1:1111")),
- taskExecutor,
- mailboxRegistry,
- messageService
- );
+ var exchangeService = new ExchangeServiceImpl(clusterNode,
taskExecutor, mailboxRegistry, messageService);
var schemaManagerMock = mock(SqlSchemaManager.class);
when(schemaManagerMock.tableById(any(), anyInt())).thenReturn(table);
var executionService = new ExecutionServiceImpl<>(
- new ClusterNode(nodeId, "fake-test-node",
NetworkAddress.from("127.0.0.1:1111")),
+ clusterNode,
messageService,
- (single, filter) -> single ?
List.of(nodeIds.get(ThreadLocalRandom.current().nextInt(nodeIds.size()))) :
nodeIds,
+ (single, filter) -> single ?
List.of(nodeNames.get(ThreadLocalRandom.current().nextInt(nodeNames.size()))) :
nodeNames,
schemaManagerMock,
mock(DdlCommandHandler.class),
taskExecutor,
@@ -403,12 +400,12 @@ public class ExecutionServiceImplTest {
static class TestCluster {
private final Map<String, TestNode> nodes = new ConcurrentHashMap<>();
- public TestNode addNode(String nodeId, QueryTaskExecutor taskExecutor)
{
- return nodes.computeIfAbsent(nodeId, key -> new TestNode(nodeId,
taskExecutor));
+ public TestNode addNode(String nodeName, QueryTaskExecutor
taskExecutor) {
+ return nodes.computeIfAbsent(nodeName, key -> new
TestNode(nodeName, taskExecutor));
}
- public TestNode node(String nodeId) {
- return nodes.get(nodeId);
+ public TestNode node(String nodeName) {
+ return nodes.get(nodeName);
}
class TestNode {
@@ -419,12 +416,12 @@ public class ExecutionServiceImplTest {
private volatile MessageInterceptor interceptor = null;
private final QueryTaskExecutor taskExecutor;
- private final String nodeId;
+ private final String nodeName;
private boolean scanPaused = false;
- public TestNode(String nodeId, QueryTaskExecutor taskExecutor) {
- this.nodeId = nodeId;
+ public TestNode(String nodeName, QueryTaskExecutor taskExecutor) {
+ this.nodeName = nodeName;
this.taskExecutor = taskExecutor;
}
@@ -473,16 +470,16 @@ public class ExecutionServiceImplTest {
return new MessageService() {
/** {@inheritDoc} */
@Override
- public void send(String targetNodeId, NetworkMessage msg) {
- TestNode node = nodes.get(targetNodeId);
+ public void send(String nodeName, NetworkMessage msg) {
+ TestNode node = nodes.get(nodeName);
- node.onReceive(nodeId, msg);
+ node.onReceive(TestNode.this.nodeName, msg);
}
/** {@inheritDoc} */
@Override
- public boolean alive(String nodeId) {
- return !nodes.get(nodeId).dead();
+ public boolean alive(String nodeName) {
+ return !nodes.get(nodeName).dead();
}
/** {@inheritDoc} */
@@ -491,7 +488,7 @@ public class ExecutionServiceImplTest {
var old = msgListeners.put(msgId, lsnr);
if (old != null) {
- throw new RuntimeException(format("Listener was
replaced [nodeId={}, msgId={}]", nodeId, msgId));
+ throw new RuntimeException(format("Listener was
replaced [nodeName={}, msgId={}]", nodeName, msgId));
}
}
@@ -539,35 +536,35 @@ public class ExecutionServiceImplTest {
};
}
- private void onReceive(String senderNodeId, NetworkMessage
message) {
- MessageListener original = (nodeId, msg) -> {
+ private void onReceive(String senderNodeName, NetworkMessage
message) {
+ MessageListener original = (nodeName, msg) -> {
MessageListener listener =
msgListeners.get(msg.messageType());
if (listener == null) {
throw new IllegalStateException(
- format("Listener not found [senderNodeId={},
msgId={}]", nodeId, msg.messageType()));
+ format("Listener not found [senderNodeName={},
msgId={}]", nodeName, msg.messageType()));
}
if (msg instanceof ExecutionContextAwareMessage) {
ExecutionContextAwareMessage msg0 =
(ExecutionContextAwareMessage) msg;
- taskExecutor.execute(msg0.queryId(),
msg0.fragmentId(), () -> listener.onMessage(nodeId, msg));
+ taskExecutor.execute(msg0.queryId(),
msg0.fragmentId(), () -> listener.onMessage(nodeName, msg));
} else {
- taskExecutor.execute(() -> listener.onMessage(nodeId,
msg));
+ taskExecutor.execute(() ->
listener.onMessage(nodeName, msg));
}
};
MessageInterceptor interceptor = this.interceptor;
if (interceptor != null) {
- interceptor.intercept(senderNodeId, message, original);
+ interceptor.intercept(senderNodeName, message, original);
} else {
- original.onMessage(senderNodeId, message);
+ original.onMessage(senderNodeName, message);
}
}
}
interface MessageInterceptor {
- void intercept(String senderNodeId, NetworkMessage msg,
MessageListener original);
+ void intercept(String senderNodeName, NetworkMessage msg,
MessageListener original);
}
}
@@ -602,7 +599,7 @@ public class ExecutionServiceImplTest {
@Override
public ColocationGroup colocationGroup(MappingQueryContext ctx) {
- return ColocationGroup.forNodes(nodeIds);
+ return ColocationGroup.forNodes(nodeNames);
}
};
}
diff --git
a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/TableScanNodeExecutionTest.java
b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/TableScanNodeExecutionTest.java
index e16becdf44..ac18d3c684 100644
---
a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/TableScanNodeExecutionTest.java
+++
b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/exec/rel/TableScanNodeExecutionTest.java
@@ -27,7 +27,6 @@ import java.util.Collection;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadLocalRandom;
-import java.util.function.Function;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.HybridClockImpl;
@@ -57,7 +56,6 @@ import org.apache.ignite.raft.client.service.RaftGroupService;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.Test;
-import org.mockito.Mockito;
/**
* Tests execution flow of TableScanNode.
@@ -126,7 +124,7 @@ public class TableScanNodeExecutionTest extends
AbstractExecutionTest {
@Override
public InternalTable table() {
- return new
TestInternalTableImpl(Mockito.mock(ReplicaService.class));
+ return new TestInternalTableImpl(mock(ReplicaService.class));
}
@Override
@@ -151,8 +149,7 @@ public class TableScanNodeExecutionTest extends
AbstractExecutionTest {
UUID.randomUUID(),
Int2ObjectMaps.singleton(0, mock(RaftGroupService.class)),
PART_CNT,
- Function.identity(),
- addr -> Mockito.mock(ClusterNode.class),
+ addr -> mock(ClusterNode.class),
new TxManagerImpl(replicaSvc, new HeapLockManager(), new
HybridClockImpl()),
mock(MvTableStorage.class),
mock(TxStateTableStorage.class),
diff --git
a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTablePersistenceTest.java
b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTablePersistenceTest.java
index b4d4255cd8..d74a09714d 100644
---
a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTablePersistenceTest.java
+++
b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTablePersistenceTest.java
@@ -130,7 +130,6 @@ public class ItTablePersistenceTest extends
ItAbstractListenerSnapshotTest<Parti
UUID.randomUUID(),
Int2ObjectMaps.singleton(0, service),
1,
- Function.identity(),
consistentIdToNode,
txManager,
mock(MvTableStorage.class),
@@ -157,7 +156,6 @@ public class ItTablePersistenceTest extends
ItAbstractListenerSnapshotTest<Parti
UUID.randomUUID(),
Int2ObjectMaps.singleton(0, service),
1,
- Function.identity(),
consistentIdToNode,
txManager,
mock(MvTableStorage.class),
@@ -190,7 +188,6 @@ public class ItTablePersistenceTest extends
ItAbstractListenerSnapshotTest<Parti
UUID.randomUUID(),
Int2ObjectMaps.singleton(0, service),
1,
- Function.identity(),
consistentIdToNode,
txManager,
mock(MvTableStorage.class),
diff --git
a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
index f6b094a324..50c6109df6 100644
---
a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
+++
b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ItTxDistributedTestSingleNode.java
@@ -333,7 +333,6 @@ public class ItTxDistributedTestSingleNode extends
TxAbstractTest {
accTblId,
accRaftClients,
1,
- Function.identity(),
consistentIdToNode,
txMgr,
Mockito.mock(MvTableStorage.class),
@@ -347,7 +346,6 @@ public class ItTxDistributedTestSingleNode extends
TxAbstractTest {
custTblId,
custRaftClients,
1,
- Function.identity(),
consistentIdToNode,
txMgr,
Mockito.mock(MvTableStorage.class),
diff --git
a/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java
b/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java
index b910c8fb6a..23312776c4 100644
---
a/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java
+++
b/modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItColocationTest.java
@@ -206,8 +206,7 @@ public class ItColocationTest {
tblId,
partRafts,
PARTS,
- null,
- address -> clusterNode,
+ name -> clusterNode,
txManager,
Mockito.mock(MvTableStorage.class),
new TestTxStateTableStorage(),
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
index 8b62691b85..ff0fd9a51d 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/TableManager.java
@@ -249,9 +249,6 @@ public class TableManager extends Producer<TableEvent,
TableEventParameters> imp
*/
private final Map<UUID, TableImpl> tablesToStopInCaseOfError = new
ConcurrentHashMap<>();
- /** Resolver that resolves a node consistent ID to node id. */
- private final Function<String, String> nodeIdResolver;
-
/** Resolver that resolves a node consistent ID to cluster node. */
private final Function<String, ClusterNode> clusterNodeResolver;
@@ -354,16 +351,6 @@ public class TableManager extends Producer<TableEvent,
TableEventParameters> imp
placementDriver = new PlacementDriver(replicaSvc);
- nodeIdResolver = consistentId -> {
- ClusterNode node = topologyService.getByConsistentId(consistentId);
-
- if (node == null) {
- throw new IllegalStateException("Can't resolve ClusterNode by
its consistent ID =" + consistentId);
- }
-
- return node.id();
- };
-
clusterNodeResolver = topologyService::getByConsistentId;
tablesByIdVv = new VersionedValue<>(null, HashMap::new);
@@ -855,8 +842,7 @@ public class TableManager extends Producer<TableEvent,
TableEventParameters> imp
txStatePartitionStorage,
topologyService,
placementDriver,
- peer
-> clusterNodeResolver.apply(peer.consistentId())
-
.equals(topologyService.localMember())
+
this::isLocalPeer
)
);
} catch
(NodeStoppingException ex) {
@@ -889,6 +875,10 @@ public class TableManager extends Producer<TableEvent,
TableEventParameters> imp
allOf(futures).join();
}
+ private boolean isLocalPeer(Peer peer) {
+ return
peer.consistentId().equals(raftMgr.topologyService().localMember().name());
+ }
+
private PartitionDataStorage partitionDataStorage(MvPartitionStorage
partitionStorage, InternalTable internalTbl, int partId) {
return new SnapshotAwarePartitionDataStorage(
partitionStorage,
@@ -1053,7 +1043,7 @@ public class TableManager extends Producer<TableEvent,
TableEventParameters> imp
TxStateTableStorage txStateStorage =
createTxStateTableStorage(tableCfg);
InternalTableImpl internalTable = new InternalTableImpl(name, tblId,
new Int2ObjectOpenHashMap<>(partitions),
- partitions, nodeIdResolver, clusterNodeResolver, txManager,
tableStorage, txStateStorage, replicaSvc, clock);
+ partitions, clusterNodeResolver, txManager, tableStorage,
txStateStorage, replicaSvc, clock);
// TODO: IGNITE-16288 directIndexIds should use async configuration API
var table = new TableImpl(internalTable, lockMgr, () ->
CompletableFuture.supplyAsync(() -> directIndexIds()));
@@ -1882,8 +1872,7 @@ public class TableManager extends Producer<TableEvent,
TableEventParameters> imp
txStatePartitionStorage,
raftMgr.topologyService(),
placementDriver,
- peer ->
clusterNodeResolver.apply(peer.consistentId())
-
.equals(raftMgr.topologyService().localMember())
+ peer -> isLocalPeer(peer)
)
);
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
index 60e5e8e332..674107ff1f 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
@@ -98,9 +98,6 @@ public class InternalTableImpl implements InternalTable {
/** Table identifier. */
private final UUID tableId;
- /** Resolver that resolves a node consistent ID to node ID. */
- private final Function<String, String> nodeIdResolver;
-
/** Resolver that resolves a node consistent ID to cluster node. */
private final Function<String, ClusterNode> clusterNodeResolver;
@@ -143,7 +140,6 @@ public class InternalTableImpl implements InternalTable {
UUID tableId,
Int2ObjectMap<RaftGroupService> partMap,
int partitions,
- Function<String, String> nodeIdResolver,
Function<String, ClusterNode> clusterNodeResolver,
TxManager txManager,
MvTableStorage tableStorage,
@@ -155,7 +151,6 @@ public class InternalTableImpl implements InternalTable {
this.tableId = tableId;
this.partitionMap = partMap;
this.partitions = partitions;
- this.nodeIdResolver = nodeIdResolver;
this.clusterNodeResolver = clusterNodeResolver;
this.txManager = txManager;
this.tableStorage = tableStorage;
@@ -996,9 +991,7 @@ public class InternalTableImpl implements InternalTable {
return partitionMap.int2ObjectEntrySet().stream()
.sorted(Comparator.comparingInt(Int2ObjectOpenHashMap.Entry::getIntKey))
.map(Map.Entry::getValue)
- .map(RaftGroupService::leader)
- .map(Peer::consistentId)
- .map(nodeIdResolver)
+ .map(service -> service.leader().consistentId())
.collect(Collectors.toList());
}
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
index 1e857f0dff..f25de52421 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java
@@ -147,8 +147,7 @@ public class DummyInternalTableImpl extends
InternalTableImpl {
UUID.randomUUID(),
Int2ObjectMaps.singleton(0, mock(RaftGroupService.class)),
1,
- Function.identity(),
- addr -> mock(ClusterNode.class),
+ name -> mock(ClusterNode.class),
txManager == null ? new TxManagerImpl(replicaSvc, new
HeapLockManager(), new HybridClockImpl()) : txManager,
mock(MvTableStorage.class),
new TestTxStateTableStorage(),
