raw

Author: Vladsz83

modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryMessageFactory.java

@@ -20,13 +20,15 @@
 import org.apache.ignite.internal.codegen.TcpDiscoveryCheckFailedMessageSerializer;
 import org.apache.ignite.internal.codegen.TcpDiscoveryClientPingRequestSerializer;
 import org.apache.ignite.internal.codegen.TcpDiscoveryClientPingResponseSerializer;
+import org.apache.ignite.internal.codegen.TcpDiscoveryNodeLeftMessageSerializer;
 import org.apache.ignite.internal.codegen.TcpDiscoveryPingRequestSerializer;
 import org.apache.ignite.internal.codegen.TcpDiscoveryPingResponseSerializer;
 import org.apache.ignite.plugin.extensions.communication.MessageFactory;
 import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryCheckFailedMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryClientPingRequest;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryClientPingResponse;
+import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeLeftMessage;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryPingRequest;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryPingResponse;
 
@@ -39,5 +41,6 @@ public class DiscoveryMessageFactory implements MessageFactoryProvider {
         factory.register((short)2, TcpDiscoveryPingResponse::new, new TcpDiscoveryPingResponseSerializer());
         factory.register((short)3, TcpDiscoveryClientPingRequest::new, new TcpDiscoveryClientPingRequestSerializer());
         factory.register((short)4, TcpDiscoveryClientPingResponse::new, new TcpDiscoveryClientPingResponseSerializer());
+        factory.register((short)5, TcpDiscoveryNodeLeftMessage::new, new TcpDiscoveryNodeLeftMessageSerializer());
     }
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/tracing/messages/SpanContainer.java

@@ -46,7 +46,7 @@ public byte[] serializedSpanBytes() {
      * @param serializedSpan Serialized span.
      */
     public void serializedSpanBytes(byte[] serializedSpan) {
-        this.serializedSpanBytes = serializedSpan.clone();
+        serializedSpanBytes = serializedSpan.clone();
     }
 
     /**

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryAbstractTraceableMessage.java

@@ -17,10 +17,12 @@
 
 package org.apache.ignite.spi.discovery.tcp.messages;
 
-import java.io.Externalizable;
 import java.util.UUID;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.discovery.DiscoveryMessageFactory;
 import org.apache.ignite.internal.processors.tracing.messages.SpanContainer;
 import org.apache.ignite.internal.processors.tracing.messages.TraceableMessage;
+import org.jetbrains.annotations.Nullable;
 
 /**
  * Abstract traceable message for TCP discovery.
@@ -29,8 +31,12 @@ public abstract class TcpDiscoveryAbstractTraceableMessage extends TcpDiscoveryA
     /** Container. */
     private SpanContainer spanContainer = new SpanContainer();
 
+    /** Serialization holder of {@link #spanContainer}'s bytes. */
+    @Order(value = 5, method = "spanBytes")
+    private @Nullable byte[] spanBytesHolder;
+
     /**
-     * Default no-arg constructor for {@link Externalizable} interface.
+     * Default constructor for {@link DiscoveryMessageFactory}.
      */
     protected TcpDiscoveryAbstractTraceableMessage() {
         // No-op.
@@ -51,7 +57,7 @@ protected TcpDiscoveryAbstractTraceableMessage(UUID creatorNodeId) {
     protected TcpDiscoveryAbstractTraceableMessage(TcpDiscoveryAbstractTraceableMessage msg) {
         super(msg);
 
-        this.spanContainer = msg.spanContainer;
+        spanContainer = msg.spanContainer;
     }
 
     /**
@@ -67,6 +73,21 @@ public Object readResolve() {
         return this;
     }
 
+    /** @return {@link #spanContainer}'s bytes. */
+    public @Nullable byte[] spanBytes() {
+        return spanContainer == null ? null : spanContainer.serializedSpanBytes();
+    }
+
+    /** @param spanBytes {@link #spanContainer}'s bytes. */
+    public void spanBytes(@Nullable byte[] spanBytes) {
+        if (spanBytes == null)
+            return;
+
+        readResolve();
+
+        spanContainer.serializedSpanBytes(spanBytes);
+    }
+
     /** {@inheritDoc} */
     @Override public SpanContainer spanContainer() {
         return spanContainer;

modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/messages/TcpDiscoveryNodeLeftMessage.java

@@ -18,18 +18,25 @@
 package org.apache.ignite.spi.discovery.tcp.messages;
 
 import java.util.UUID;
+import org.apache.ignite.internal.managers.discovery.DiscoveryMessageFactory;
 import org.apache.ignite.internal.util.typedef.internal.S;
+import org.apache.ignite.plugin.extensions.communication.Message;
 
 /**
  * Sent by node that is stopping to coordinator across the ring,
  * then sent by coordinator across the ring.
  */
 @TcpDiscoveryEnsureDelivery
 @TcpDiscoveryRedirectToClient
-public class TcpDiscoveryNodeLeftMessage extends TcpDiscoveryAbstractTraceableMessage {
+public class TcpDiscoveryNodeLeftMessage extends TcpDiscoveryAbstractTraceableMessage implements Message {
     /** */
     private static final long serialVersionUID = 0L;
 
+    /** Default constructor for {@link DiscoveryMessageFactory}. */
+    public TcpDiscoveryNodeLeftMessage() {
+        // No-op.
+    }
+
     /**
      * Constructor.
      *
@@ -43,4 +50,9 @@ public TcpDiscoveryNodeLeftMessage(UUID creatorNodeId) {
     @Override public String toString() {
         return S.toString(TcpDiscoveryNodeLeftMessage.class, this, "super", super.toString());
     }
+
+    /** */
+    @Override public short directType() {
+        return 5;
+    }
 }