Merge remote-tracking branch 'origin/master' into ignite-26584

Author: anton-vinogradov

modules/calcite/src/test/java/org/apache/ignite/internal/processors/query/calcite/message/CalciteCommunicationMessageSerializationTest.java

@@ -17,11 +17,11 @@
 
 package org.apache.ignite.internal.processors.query.calcite.message;
 
-import org.apache.ignite.internal.managers.communication.AbstractCommunicationMessageSerializationTest;
+import org.apache.ignite.internal.managers.communication.AbstractMessageSerializationTest;
 import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
 
 /** */
-public class CalciteCommunicationMessageSerializationTest extends AbstractCommunicationMessageSerializationTest {
+public class CalciteCommunicationMessageSerializationTest extends AbstractMessageSerializationTest {
     /** {@inheritDoc} */
     @Override protected MessageFactoryProvider messageFactory() {
         return new CalciteMessageFactory();

modules/compress/pom.xml

@@ -67,7 +67,7 @@
         </dependency>
 
         <dependency>
-            <groupId>org.lz4</groupId>
+            <groupId>at.yawk.lz4</groupId>
             <artifactId>lz4-java</artifactId>
             <version>${lz4.version}</version>
         </dependency>

modules/core/src/main/java/org/apache/ignite/internal/IgniteNodeAttributes.java

@@ -150,6 +150,9 @@ public final class IgniteNodeAttributes {
     /** V2 security subject for authenticated node. */
     public static final String ATTR_SECURITY_SUBJECT_V2 = ATTR_PREFIX + ".security.subject.v2";
 
+    /** Node certificates the connection was established with.  */
+    public static final String ATTR_NODE_CERTIFICATES = ATTR_PREFIX + ".security.certificates";
+
     /** Client mode flag. */
     public static final String ATTR_CLIENT_MODE = ATTR_PREFIX + ".cache.client";
 

modules/core/src/main/java/org/apache/ignite/internal/client/thin/TcpClientCache.java

@@ -71,6 +71,7 @@
 import org.apache.ignite.internal.processors.cache.CacheInvokeResult;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.processors.platform.client.ClientStatus;
+import org.apache.ignite.internal.util.typedef.F;
 import org.apache.ignite.internal.util.typedef.T3;
 import org.apache.ignite.internal.util.typedef.X;
 import org.apache.ignite.internal.util.typedef.internal.A;
@@ -1389,7 +1390,7 @@ private QueryCursor<Cache.Entry<K, V>> sqlQuery(SqlQuery qry) {
             keepBinary,
             marsh,
             cacheId,
-            qry.getPartitions().length >= 1 ? qry.getPartitions()[0] : -1
+            !F.isEmpty(qry.getPartitions()) ? qry.getPartitions()[0] : -1
         ));
     }
 

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

@@ -18,11 +18,15 @@
 package org.apache.ignite.internal.managers.discovery;
 
 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.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.TcpDiscoveryPingRequest;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryPingResponse;
 
@@ -33,5 +37,7 @@ public class DiscoveryMessageFactory implements MessageFactoryProvider {
         factory.register((short)0, TcpDiscoveryCheckFailedMessage::new, new TcpDiscoveryCheckFailedMessageSerializer());
         factory.register((short)1, TcpDiscoveryPingRequest::new, new TcpDiscoveryPingRequestSerializer());
         factory.register((short)2, TcpDiscoveryPingResponse::new, new TcpDiscoveryPingResponseSerializer());
+        factory.register((short)3, TcpDiscoveryClientPingRequest::new, new TcpDiscoveryClientPingRequestSerializer());
+        factory.register((short)4, TcpDiscoveryClientPingResponse::new, new TcpDiscoveryClientPingResponseSerializer());
     }
 }

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java

@@ -1577,7 +1577,7 @@ private void finishIncrementalSnapshotRestore(UUID reqId, Map<UUID, Boolean> res
      * @param metas Map of snapshot metadata distribution across the cluster.
      * @return Map of cache partitions per each node.
      */
-    private static Map<UUID, Map<Integer, Set<Integer>>> snapshotAffinity(
+    private Map<UUID, Map<Integer, Set<Integer>>> snapshotAffinity(
         Map<UUID, List<SnapshotMetadata>> metas,
         BiPredicate<Integer, Integer> filter
     ) {
@@ -1586,10 +1586,23 @@ private static Map<UUID, Map<Integer, Set<Integer>>> snapshotAffinity(
         List<UUID> nodes = new ArrayList<>(metas.keySet());
         Collections.shuffle(nodes);
 
-        Map<UUID, List<SnapshotMetadata>> shuffleMetas = new LinkedHashMap<>();
-        nodes.forEach(k -> shuffleMetas.put(k, metas.get(k)));
+        Map<UUID, List<SnapshotMetadata>> orderedMetas = new LinkedHashMap<>();
 
-        for (Map.Entry<UUID, List<SnapshotMetadata>> e : shuffleMetas.entrySet()) {
+        String locDc = ctx.discovery().localNode().dataCenterId();
+
+        if (locDc != null) {
+            List<UUID> sameDcNodes = nodes.stream()
+                .map(uuid -> ctx.discovery().node(uuid))
+                .filter(node -> Objects.equals(node.dataCenterId(), locDc))
+                .map(ClusterNode::id)
+                .collect(Collectors.toList());
+
+            sameDcNodes.forEach(k -> orderedMetas.put(k, metas.get(k))); // Getting same DC files first.
+        }
+
+        nodes.forEach(k -> orderedMetas.put(k, metas.get(k)));
+
+        for (Map.Entry<UUID, List<SnapshotMetadata>> e : orderedMetas.entrySet()) {
             UUID nodeId = e.getKey();
 
             for (SnapshotMetadata meta : ofNullable(e.getValue()).orElse(Collections.emptyList())) {

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

@@ -2301,6 +2301,9 @@ private void processNodeAddFinishedMessage(TcpDiscoveryNodeAddFinishedMessage ms
                     }
 
                     locNode.setAttributes(msg.clientNodeAttributes());
+
+                    clearNodeSensitiveData(locNode);
+
                     locNode.visible(true);
 
                     long topVer = msg.topologyVersion();
@@ -2356,6 +2359,8 @@ else if (log.isDebugEnabled())
                     assert topVer > 0 : msg;
 
                     if (!node.visible()) {
+                        clearNodeSensitiveData(node);
+
                         node.order(topVer);
                         node.visible(true);
 

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

@@ -178,6 +178,7 @@
 import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_MARSHALLER_COMPACT_FOOTER;
 import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_MARSHALLER_USE_BINARY_STRING_SER_VER_2;
 import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_MARSHALLER_USE_DFLT_SUID;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_NODE_CERTIFICATES;
 import static org.apache.ignite.internal.cluster.DistributedConfigurationUtils.CONN_DISABLED_BY_ADMIN_ERR_MSG;
 import static org.apache.ignite.internal.cluster.DistributedConfigurationUtils.newConnectionEnabledProperty;
 import static org.apache.ignite.internal.processors.security.SecurityUtils.authenticateLocalNode;
@@ -1134,7 +1135,7 @@ private void joinTopology() throws IgniteSpiException {
                         leavingNodes.clear();
                         failedNodesMsgSent.clear();
 
-                        locNode.attributes().remove(IgniteNodeAttributes.ATTR_SECURITY_CREDENTIALS);
+                        clearNodeSensitiveData(locNode);
 
                         locNode.order(1);
                         locNode.internalOrder(1);
@@ -2443,6 +2444,18 @@ private void processMessageFailedNodes(TcpDiscoveryAbstractMessage msg) {
         }
     }
 
+    /** */
+    private static void enrichNodeWithAttribute(TcpDiscoveryNode node, String attrName, @Nullable Object attrVal) {
+        if (attrVal == null)
+            return;
+
+        Map<String, Object> attrs = new HashMap<>(node.getAttributes());
+
+        attrs.put(attrName, attrVal);
+
+        node.setAttributes(attrs);
+    }
+
     /** */
     private static WorkersRegistry getWorkerRegistry(TcpDiscoverySpi spi) {
         return spi.ignite() instanceof IgniteEx ? ((IgniteEx)spi.ignite()).context().workersRegistry() : null;
@@ -5292,6 +5305,8 @@ private void processNodeAddFinishedMessage(TcpDiscoveryNodeAddFinishedMessage ms
             if (msg.verified()) {
                 assert topVer > 0 : "Invalid topology version: " + msg;
 
+                clearNodeSensitiveData(node);
+
                 if (node.order() == 0)
                     node.order(topVer);
 
@@ -7066,6 +7081,11 @@ else if (e.hasCause(ObjectStreamException.class) ||
                         else if (msg instanceof TcpDiscoveryJoinRequestMessage) {
                             TcpDiscoveryJoinRequestMessage req = (TcpDiscoveryJoinRequestMessage)msg;
 
+                            // Current node holds connection with the node that is joining the cluster. Therefore, it can
+                            // save certificates with which the connection was established to joining node attributes.
+                            if (spi.nodeAuth != null && nodeId.equals(req.node().id()))
+                                enrichNodeWithAttribute(req.node(), ATTR_NODE_CERTIFICATES, ses.extractCertificates());
+
                             if (!req.responded()) {
                                 boolean ok = processJoinRequestMessage(req, clientMsgWrk);
 

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

@@ -25,6 +25,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
@@ -53,6 +54,8 @@
 
 import static org.apache.ignite.IgniteSystemProperties.IGNITE_DISCOVERY_METRICS_QNT_WARN;
 import static org.apache.ignite.IgniteSystemProperties.getInteger;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_NODE_CERTIFICATES;
+import static org.apache.ignite.internal.IgniteNodeAttributes.ATTR_SECURITY_CREDENTIALS;
 import static org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi.DFLT_DISCOVERY_METRICS_QNT_WARN;
 
 /**
@@ -401,6 +404,16 @@ protected boolean checkAckTimeout(long ackTimeout) {
         return true;
     }
 
+    /** */
+    protected void clearNodeSensitiveData(TcpDiscoveryNode node) {
+        Map<String, Object> attrs = new HashMap<>(node.attributes());
+
+        attrs.remove(ATTR_NODE_CERTIFICATES);
+        attrs.remove(ATTR_SECURITY_CREDENTIALS);
+
+        node.setAttributes(attrs);
+    }
+
     /** */
     public void processMsgCacheMetrics(TcpDiscoveryMetricsUpdateMessage msg, long tsNanos) {
         for (Map.Entry<UUID, TcpDiscoveryMetricsUpdateMessage.MetricsSet> e : msg.metrics().entrySet()) {

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

@@ -27,6 +27,9 @@
 import java.io.StreamCorruptedException;
 import java.net.Socket;
 import java.nio.ByteBuffer;
+import java.security.cert.Certificate;
+import javax.net.ssl.SSLPeerUnverifiedException;
+import javax.net.ssl.SSLSocket;
 import org.apache.ignite.IgniteCheckedException;
 import org.apache.ignite.IgniteException;
 import org.apache.ignite.internal.direct.DirectMessageReader;
@@ -36,6 +39,7 @@
 import org.apache.ignite.plugin.extensions.communication.Message;
 import org.apache.ignite.plugin.extensions.communication.MessageSerializer;
 import org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
+import org.jetbrains.annotations.Nullable;
 
 import static org.apache.ignite.spi.communication.tcp.TcpCommunicationSpi.makeMessageType;
 
@@ -206,6 +210,21 @@ <T> T readMessage() throws IgniteCheckedException, IOException {
         }
     }
 
+    /** @return SSL certificate this session is established with. {@code null} if SSL is disabled or certificate validation failed. */
+    @Nullable Certificate[] extractCertificates() {
+        if (!spi.isSslEnabled())
+            return null;
+
+        try {
+            return ((SSLSocket)sock).getSession().getPeerCertificates();
+        }
+        catch (SSLPeerUnverifiedException e) {
+            U.error(spi.log, "Failed to extract discovery IO session certificates", e);
+
+            return null;
+        }
+    }
+
     /**
      * Serializes a discovery message into a byte array.
      *