Skip to content

IGNITE-29037 Refactored Message Factory usage in TCP Discovery implementations #13551

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
petrov-mg merged 2 commits into from
Sep 4, 2026
+85 −188
Merged

IGNITE-29037 Refactored Message Factory usage in TCP Discovery implementations #13551

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
2810296
IGNITE-29037
petrov-mg Sep 3, 2026
957c5d0
IGNITE-29037
petrov-mg Sep 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
32 changes: 10 additions & 22 deletions modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ClientImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,6 @@
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.atomic.AtomicReference;
import javax.net.ssl.SSLException;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.IgniteClientDisconnectedException;
import org.apache.ignite.IgniteException;
Expand All @@ -60,7 +59,6 @@
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.failure.FailureContext;
import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.IgniteNodeAttributes;
import org.apache.ignite.internal.managers.discovery.DiscoveryServerOnlyCustomMessage;
Expand All @@ -73,7 +71,6 @@
import org.apache.ignite.internal.util.typedef.internal.LT;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.util.worker.GridWorker;
import org.apache.ignite.internal.worker.WorkersRegistry;
import org.apache.ignite.lang.IgniteInClosure;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.spi.IgniteSpiAdapter;
Expand Down Expand Up @@ -208,10 +205,9 @@ class ClientImpl extends TcpDiscoveryImpl {
ClientImpl(TcpDiscoverySpi adapter) {
super(adapter);

String instanceName = adapter.ignite() == null || adapter.ignite().name() == null
? "client-node" : adapter.ignite().name();
String instanceName = ctx.igniteInstanceName();

executorSrvc = newSingleThreadScheduledExecutor("tcp-discovery-exec", instanceName);
executorSrvc = newSingleThreadScheduledExecutor("tcp-discovery-exec", instanceName == null ? "client-node" : instanceName);
}

/** {@inheritDoc} */
Expand Down Expand Up @@ -1052,13 +1048,6 @@ private void joinError(IgniteSpiException err) {
joinLatch.countDown();
}

/** */
private WorkersRegistry getWorkersRegistry() {
Ignite ignite = spi.ignite();

return ignite instanceof IgniteEx ? ((IgniteEx)ignite).context().workersRegistry() : null;
}

/** */
private Collection<ClusterNode> remoteVisibleNodes() {
return U.arrayList(rmtNodes.values(), TcpDiscoveryNodesRing.VISIBLE_NODES);
Expand Down Expand Up @@ -1101,7 +1090,7 @@ private class SocketReader extends IgniteSpiThread {
/**
*/
SocketReader() {
super(spi.ignite().name(), "tcp-client-disco-sock-reader-[]", log);
super(ctx.igniteInstanceName(), "tcp-client-disco-sock-reader-[]", log);
}

/**
Expand Down Expand Up @@ -1274,7 +1263,7 @@ private class SocketWriter extends IgniteSpiThread {
*
*/
SocketWriter() {
super(spi.ignite().name(), "tcp-client-disco-sock-writer", log);
super(ctx.igniteInstanceName(), "tcp-client-disco-sock-writer", log);

sockTimeout = spi.failureDetectionTimeoutEnabled() ? spi.failureDetectionTimeout() :
spi.getSocketTimeout();
Expand Down Expand Up @@ -1524,7 +1513,7 @@ private class Reconnector extends IgniteSpiThread {
* @param prevAddr Address of the node, that this client was previously connected to.
*/
protected Reconnector(boolean join, InetSocketAddress prevAddr) {
super(spi.ignite().name(), "tcp-client-disco-reconnector", log);
super(ctx.igniteInstanceName(), "tcp-client-disco-reconnector", log);

this.join = join;
this.prevAddr = prevAddr;
Expand Down Expand Up @@ -1689,7 +1678,7 @@ protected class MessageWorker extends GridWorker {
* @param log Logger.
*/
private MessageWorker(IgniteLogger log) {
super(spi.ignite().name(), "tcp-client-disco-msg-worker", log, getWorkersRegistry());
super(ctx.igniteInstanceName(), "tcp-client-disco-msg-worker", log, ctx.workersRegistry());
}

/** {@inheritDoc} */
Expand Down Expand Up @@ -1957,8 +1946,7 @@ else if (discoMsg instanceof TcpDiscoveryCheckFailedMessage)
Thread.currentThread().interrupt();
}
catch (Throwable t) {
if (spi.ignite() instanceof IgniteEx)
((IgniteEx)spi.ignite()).context().failure().process(new FailureContext(CRITICAL_ERROR, t));
ctx.failure().process(new FailureContext(CRITICAL_ERROR, t));
}
finally {
TcpDiscoveryIoSession ses = this.currSes;
Expand Down Expand Up @@ -2210,7 +2198,7 @@ else if (log.isDebugEnabled())
if (joining())
delayDiscoData.add(dataPacket);
else
spi.onExchange(dataPacket, U.resolveClassLoader(spi.ignite().configuration()));
spi.onExchange(dataPacket, U.resolveClassLoader(ctx.config()));
}
}
}
Expand Down Expand Up @@ -2244,11 +2232,11 @@ private void processNodeAddFinishedMessage(TcpDiscoveryNodeAddFinishedMessage ms
DiscoveryDataPacket dataContainer = msg.clientDiscoData();

if (dataContainer != null)
spi.onExchange(dataContainer, U.resolveClassLoader(spi.ignite().configuration()));
spi.onExchange(dataContainer, U.resolveClassLoader(ctx.config()));

if (!delayDiscoData.isEmpty()) {
for (DiscoveryDataPacket data : delayDiscoData)
spi.onExchange(data, U.resolveClassLoader(spi.ignite().configuration()));
spi.onExchange(data, U.resolveClassLoader(ctx.config()));

delayDiscoData.clear();
}
Expand Down
80 changes: 32 additions & 48 deletions modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -76,7 +76,6 @@
import org.apache.ignite.events.NodeValidationFailedEvent;
import org.apache.ignite.failure.FailureContext;
import org.apache.ignite.internal.ClusterMetricsSnapshot;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteFutureTimeoutCheckedException;
import org.apache.ignite.internal.IgniteInterruptedCheckedException;
import org.apache.ignite.internal.IgniteNodeAttributes;
Expand Down Expand Up @@ -107,7 +106,6 @@
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.internal.util.worker.GridWorker;
import org.apache.ignite.internal.util.worker.GridWorkerListener;
import org.apache.ignite.internal.worker.WorkersRegistry;
import org.apache.ignite.lang.IgniteBiTuple;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteInClosure;
Expand Down Expand Up @@ -362,14 +360,14 @@ class ServerImpl extends TcpDiscoveryImpl {
super(adapter);

utilityPool = new IgniteThreadPoolExecutor("disco-pool",
spi.ignite().name(),
ctx.igniteInstanceName(),
0,
utilityPoolSize,
2000,
new LinkedBlockingQueue<>());

List<DistributedBooleanProperty> props = newConnectionEnabledProperty(
((IgniteEx)spi.ignite()).context().internalSubscriptionProcessor(),
ctx.internalSubscriptionProcessor(),
log,
"ClientNode",
"ServerNode"
Expand Down Expand Up @@ -2249,7 +2247,7 @@ private class IpFinderCleaner extends IgniteSpiThread {
* Constructor.
*/
private IpFinderCleaner() {
super(spi.ignite().name(), "tcp-disco-ip-finder-cleaner", log);
super(ctx.igniteInstanceName(), "tcp-disco-ip-finder-cleaner", log);

setPriority(spi.threadPri);
}
Expand Down Expand Up @@ -2458,11 +2456,6 @@ private static void enrichNodeWithAttribute(TcpDiscoveryNode node, String attrNa
node.setAttributes(attrs);
}

/** */
private static WorkersRegistry getWorkerRegistry(TcpDiscoverySpi spi) {
return spi.ignite() instanceof IgniteEx ? ((IgniteEx)spi.ignite()).context().workersRegistry() : null;
}

/**
* Upcasts collection type.
*
Expand Down Expand Up @@ -2866,7 +2859,7 @@ protected class RingMessageWorker extends MessageWorker<TcpDiscoveryAbstractMess
// To address this, we use TcpDiscoveryMessageSerializer, which includes some code copied from TcpDiscoveryIoSession
// and can be instantiated independently of any active session.
/** */
private final TcpDiscoveryMessageSerializer clientMsgSer = new TcpDiscoveryMessageSerializer(spi);
private final TcpDiscoveryMessageSerializer clientMsgSer = new TcpDiscoveryMessageSerializer(ctx);

/** IO session. */
private TcpDiscoveryIoSession ses;
Expand Down Expand Up @@ -2902,7 +2895,7 @@ protected class RingMessageWorker extends MessageWorker<TcpDiscoveryAbstractMess

/** */
protected RingMessageWorker(IgniteLogger log, BlockingDeque<TcpDiscoveryAbstractMessage> queue) {
super("tcp-disco-msg-worker-[]", log, 10, getWorkerRegistry(spi), queue);
super("tcp-disco-msg-worker-[]", log, 10, ctx.workersRegistry(), queue);

setBeforeEachPollAction(() -> {
updateHeartbeat();
Expand Down Expand Up @@ -3049,17 +3042,15 @@ private void addToQueue(TcpDiscoveryAbstractMessage msg, boolean addFirst) {
throw e;
}
finally {
if (spi.ignite() instanceof IgniteEx) {
if (err == null && !spi.isNodeStopping0() && spiStateCopy() != DISCONNECTING)
err = new IllegalStateException("Worker " + name() + " is terminated unexpectedly.");
if (err == null && !spi.isNodeStopping0() && spiStateCopy() != DISCONNECTING)
err = new IllegalStateException("Worker " + name() + " is terminated unexpectedly.");

FailureProcessor failure = ((IgniteEx)spi.ignite()).context().failure();
FailureProcessor failure = ctx.failure();

if (err instanceof OutOfMemoryError)
failure.process(new FailureContext(CRITICAL_ERROR, err));
else if (err != null)
failure.process(new FailureContext(SYSTEM_WORKER_TERMINATION, err));
}
if (err instanceof OutOfMemoryError)
failure.process(new FailureContext(CRITICAL_ERROR, err));
else if (err != null)
failure.process(new FailureContext(SYSTEM_WORKER_TERMINATION, err));
}
}

Expand Down Expand Up @@ -4591,8 +4582,7 @@ private IgniteNodeValidationResult validateByIgniteComponentsWithJoiningNodeData
DiscoveryDataPacket packet = req.gridDiscoveryData();

try {
DiscoveryDataBag dataBag = packet.bagWithJoiningNodeData(spi.ignite().log(),
spi.ignite().configuration().isClientMode());
DiscoveryDataBag dataBag = packet.bagWithJoiningNodeData(log, ctx.config().isClientMode());

return spi.getSpiContext().validateNode(req.node(), dataBag);
}
Expand Down Expand Up @@ -4936,7 +4926,7 @@ else if (!locNodeId.equals(node.id()) && ring.node(node.id()) != null) {
if (dataPacket.hasJoiningNodeData()) {
if (spiState == CONNECTED) {
// Node already connected to the cluster can apply joining nodes' disco data immediately
spi.onExchange(dataPacket, U.resolveClassLoader(spi.ignite().configuration()));
spi.onExchange(dataPacket, U.resolveClassLoader(ctx.config()));

spi.collectExchangeData(dataPacket);
}
Expand Down Expand Up @@ -5154,11 +5144,11 @@ private void processNodeAddFinishedMessage(TcpDiscoveryNodeAddFinishedMessage ms
}

if (gridDiscoveryData != null)
spi.onExchange(gridDiscoveryData, U.resolveClassLoader(spi.ignite().configuration()));
spi.onExchange(gridDiscoveryData, U.resolveClassLoader(ctx.config()));

if (joiningNodesDiscoDataList != null) {
for (DiscoveryDataPacket dataPacket : joiningNodesDiscoDataList)
spi.onExchange(dataPacket, U.resolveClassLoader(spi.ignite().configuration()));
spi.onExchange(dataPacket, U.resolveClassLoader(ctx.config()));
}

nullifyDiscoData();
Expand Down Expand Up @@ -6310,7 +6300,7 @@ private class TcpServer extends GridWorker {
* @throws IgniteSpiException In case of error.
*/
TcpServer(IgniteLogger log) throws IgniteSpiException {
super(spi.ignite().name(), "tcp-disco-srvr-[]", log, getWorkerRegistry(spi));
super(ctx.igniteInstanceName(), "tcp-disco-srvr-[]", log, ctx.workersRegistry());

int lastPort = spi.locPortRange == 0 ? spi.locPort : spi.locPort + spi.locPortRange - 1;

Expand All @@ -6330,7 +6320,7 @@ private class TcpServer extends GridWorker {
if (log.isInfoEnabled()) {
log.info("Successfully bound to TCP port [port=" + port +
", localHost=" + spi.locHost +
", locNodeId=" + spi.ignite().configuration().getNodeId() +
", locNodeId=" + spi.cfgNodeId +
']');
}

Expand Down Expand Up @@ -6413,17 +6403,15 @@ private class TcpServer extends GridWorker {
throw t;
}
finally {
if (spi.ignite() instanceof IgniteEx) {
if (err == null && !spi.isNodeStopping0() && spiStateCopy() != DISCONNECTING)
err = new IllegalStateException("Worker " + name() + " is terminated unexpectedly.");
if (err == null && !spi.isNodeStopping0() && spiStateCopy() != DISCONNECTING)
err = new IllegalStateException("Worker " + name() + " is terminated unexpectedly.");

FailureProcessor failure = ((IgniteEx)spi.ignite()).context().failure();
FailureProcessor failure = ctx.failure();

if (err instanceof OutOfMemoryError)
failure.process(new FailureContext(CRITICAL_ERROR, err));
else if (err != null)
failure.process(new FailureContext(SYSTEM_WORKER_TERMINATION, err));
}
if (err instanceof OutOfMemoryError)
failure.process(new FailureContext(CRITICAL_ERROR, err));
else if (err != null)
failure.process(new FailureContext(SYSTEM_WORKER_TERMINATION, err));

U.closeQuiet(srvrSock);
}
Expand Down Expand Up @@ -6458,11 +6446,11 @@ private class SocketReader extends IgniteSpiThread {
* @param sock Socket to read data from.
*/
SocketReader(Socket sock) {
super(spi.ignite().name(), "tcp-disco-sock-reader-[]", log);
super(ctx.igniteInstanceName(), "tcp-disco-sock-reader-[]", log);

this.sock = sock;

ses = createSession(sock);
ses = new TcpDiscoveryIoSession(ctx, sock);

setPriority(spi.threadPri);
}
Expand Down Expand Up @@ -7085,11 +7073,7 @@ else if (msg instanceof TcpDiscoveryRingLatencyCheckMessage) {
}
}
catch (UnknownMessageException e) {
if (spi.ignite() instanceof IgniteEx) {
FailureProcessor failure = ((IgniteEx)spi.ignite()).context().failure();

failure.process(new FailureContext(SYSTEM_WORKER_TERMINATION, e));
}
ctx.failure().process(new FailureContext(SYSTEM_WORKER_TERMINATION, e));
}
finally {
if (clientMsgWrk != null) {
Expand Down Expand Up @@ -7461,7 +7445,7 @@ private class StatisticsPrinter extends IgniteSpiThread {
* Constructor.
*/
StatisticsPrinter() {
super(spi.ignite().name(), "tcp-disco-stats-printer", log);
super(ctx.igniteInstanceName(), "tcp-disco-stats-printer", log);

assert spi.statsPrintFreq > 0;

Expand Down Expand Up @@ -7534,7 +7518,7 @@ private ClientMessageWorker(TcpDiscoveryIoSession ses, UUID clientNodeId, Ignite
this.ses = ses;
this.clientNodeId = clientNodeId;

clientMsgSer = new TcpDiscoveryMessageSerializer(spi);
clientMsgSer = new TcpDiscoveryMessageSerializer(ctx);

lastMetricsUpdateMsgTimeNanos = System.nanoTime();
}
Expand Down Expand Up @@ -7878,7 +7862,7 @@ protected MessageWorker(
@Nullable GridWorkerListener lsnr,
BlockingDeque<T> queue
) {
super(spi.ignite().name(), name, log, lsnr);
super(ctx.igniteInstanceName(), name, log, lsnr);

this.queue = queue;
this.pollingTimeout = pollingTimeout;
Expand Down Expand Up @@ -8101,7 +8085,7 @@ void pingRemoteDCs(List<TcpDiscoveryNode> nodesToPing) {

rmtDcPingPool = new IgniteThreadPoolExecutor(
"disco-remote-dc-ping-worker",
spi.ignite().name(),
ctx.igniteInstanceName(),
pingRmtDcPoolSz,
pingRmtDcPoolSz,
0,
Expand Down
Loading
Loading