diff --git a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java index e412de189b416..d95dbd5c207c1 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java @@ -220,6 +220,9 @@ import org.apache.ignite.internal.processors.marshaller.MissingMappingResponseMessage; import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageCasAckMessage; import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageCasMessage; +import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageClusterNodeData; +import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageHistoryItemMessage; +import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageJoiningNodeData; import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageUpdateAckMessage; import org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageUpdateMessage; import org.apache.ignite.internal.processors.plugin.PluginsDataBagItem; @@ -727,6 +730,9 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh) { register(BaselineTopologyHistory.class); register(BaselineTopologyHistoryItem.class); register(DiscoveryDataClusterState.class); + register(DistributedMetaStorageHistoryItemMessage.class); + register(DistributedMetaStorageJoiningNodeData.class); + register(DistributedMetaStorageClusterNodeData.class); // [13400 - 13500]: Operation context messages. msgIdx = 13400; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageClusterNodeData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageClusterNodeData.java index 01375e055af15..5ccf284b8543c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageClusterNodeData.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageClusterNodeData.java @@ -17,51 +17,85 @@ package org.apache.ignite.internal.processors.metastorage.persistence; -import java.io.Serializable; +import java.io.Externalizable; +import java.util.Map; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage; +import org.apache.ignite.internal.util.tostring.GridToStringInclude; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.jetbrains.annotations.Nullable; /** - * Distributed metastorage data that cluster sends to joining node. + * Distributed metastorage data that cluster sends to joining node. To reduce messages number, contains plain representation + * of {@link DistributedMetaStorageVersion}, arrays of plain representations of Distributed MetaStorage's key-value pairs. + * And wrapped {@link DistributedMetaStorageHistoryItem}s. The version and the full data holders are {@link Externalizable}s + * persistent by {@link MetaStorage} with the dedicated code-generated serializers. Thus, we do not make them directly a {@link Message}. + * + * @see DmsDataWriter#write(String, byte[]) + * @see MetaStorage#write(String, Serializable) */ -@SuppressWarnings({"PublicField", "AssignmentOrReturnOfFieldWithMutableType"}) -class DistributedMetaStorageClusterNodeData implements Serializable { - /** */ - private static final long serialVersionUID = 0L; - - /** - * Distributed metastorage version of cluster. If {@link #fullData} is not null then this version corresponds to - * its content. - */ - public final DistributedMetaStorageVersion ver; - - /** - * Full data is sent if there's not enough history items on local node. - */ - public final DistributedMetaStorageKeyValuePair[] fullData; - - /** - * Required updates for joining nodes or full available history of local node if {@link #fullData} is - * not {@code null}. - */ - public final DistributedMetaStorageHistoryItem[] hist; - - /** - * Additional updates. Makes sence only if {@link #fullData} is not {@code null}. - */ - public DistributedMetaStorageHistoryItem[] updates; +public class DistributedMetaStorageClusterNodeData implements Message { + /** @see DistributedMetaStorageVersion#id */ + @Order(0) + @GridToStringInclude + long dVerId; + + /** @see DistributedMetaStorageVersion#hash */ + @Order(1) + @GridToStringInclude + long dVerHash; + + /** Array of the full data keys. */ + @GridToStringInclude + @Order(2) + @Nullable String[] fullDataKeys; + + /** Arrays of the full data bytes. */ + @GridToStringInclude + @Order(3) + @Nullable byte[][] fullDataValsBytes; + + /** Required updates for joining nodes or full available history of local node if the full data is not {@code null}. */ + @Order(4) + @Nullable DistributedMetaStorageHistoryItemMessage[] hist; + + /** Additional updates. Makes sense only if the full data is not {@code null}. */ + @Order(5) + @Nullable DistributedMetaStorageHistoryItemMessage[] updates; + + /** Empty constructor for serialization purposes. */ + public DistributedMetaStorageClusterNodeData() { + // No-op. + } /** */ public DistributedMetaStorageClusterNodeData( DistributedMetaStorageVersion ver, - DistributedMetaStorageKeyValuePair[] fullData, - DistributedMetaStorageHistoryItem[] hist, - DistributedMetaStorageHistoryItem[] updates + @Nullable Map fullData, + @Nullable DistributedMetaStorageHistoryItem[] hist, + @Nullable DistributedMetaStorageHistoryItem[] updates ) { assert ver != null; assert fullData == null || hist != null; - this.fullData = fullData; - this.ver = ver; - this.hist = hist; - this.updates = updates; + dVerId = ver.id; + dVerHash = ver.hash; + + if (fullData != null) { + fullDataKeys = new String[fullData.size()]; + fullDataValsBytes = new byte[fullData.size()][]; + + int i = 0; + + for (var e : fullData.entrySet()) { + fullDataKeys[i] = e.getKey(); + fullDataValsBytes[i] = e.getValue(); + + ++i; + } + } + + this.hist = DistributedMetaStorageHistoryItemMessage.toMessages(hist); + this.updates = DistributedMetaStorageHistoryItemMessage.toMessages(updates); } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageHistoryItem.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageHistoryItem.java index b24a275e64b53..7f05c90eb0603 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageHistoryItem.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageHistoryItem.java @@ -19,10 +19,19 @@ import java.util.Arrays; import org.apache.ignite.internal.dto.IgniteDataTransferObject; +import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage; import org.apache.ignite.internal.util.tostring.GridToStringInclude; import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.plugin.extensions.communication.Message; -/** */ +/** + * Distributed Metastorage history items holder. Is a persistent {@link IgniteDataTransferObject} stored by {@link MetaStorage} + * using the dedicated code-generated DTO-serializer. Then, has a transfer wrap {@link DistributedMetaStorageHistoryItemMessage}. + * + * @see DistributedMetaStorageHistoryItemMessage + * @see DmsDataWriter#write(String, byte[]) + * @see MetaStorage#write(String, Serializable) + */ final class DistributedMetaStorageHistoryItem extends IgniteDataTransferObject { /** */ private static final long serialVersionUID = 0L; @@ -62,6 +71,16 @@ public DistributedMetaStorageHistoryItem(String[] keys, byte[][] valBytesArr) { this.valBytesArr = valBytesArr; } + /** @return Array of {@link DistributedMetaStorageHistoryItem} created of the related {@link Message} transfer wraps. */ + static DistributedMetaStorageHistoryItem[] fromMessages(DistributedMetaStorageHistoryItemMessage[] histMsgs) { + DistributedMetaStorageHistoryItem[] res = new DistributedMetaStorageHistoryItem[histMsgs.length]; + + for (int i = 0; i < histMsgs.length; ++i) + res[i] = new DistributedMetaStorageHistoryItem(histMsgs[i].keys, histMsgs[i].valBytes); + + return res; + } + /** */ public long estimateSize() { int len = keys.length; diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageHistoryItemMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageHistoryItemMessage.java new file mode 100644 index 0000000000000..838dc2505463c --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageHistoryItemMessage.java @@ -0,0 +1,67 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.ignite.internal.processors.metastorage.persistence; + +import java.io.Serializable; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.dto.IgniteDataTransferObject; +import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.jetbrains.annotations.Nullable; + +/** + * Transfer wrap for {@link DistributedMetaStorageHistoryItem} which is a persistent {@link IgniteDataTransferObject} stored + * by {@link MetaStorage} using the dedicated code-generated DTO-serializer. + * + * @see DistributedMetaStorageHistoryItem + * @see DmsDataWriter#write(String, byte[]) + * @see MetaStorage#write(String, Serializable) + */ +public class DistributedMetaStorageHistoryItemMessage implements Message { + /** */ + @Order(0) + String[] keys; + + /** */ + @Order(1) + byte[][] valBytes; + + /** Empty constructor for serialization purposes. */ + public DistributedMetaStorageHistoryItemMessage() { + // No-op. + } + + /** */ + DistributedMetaStorageHistoryItemMessage(String[] keys, byte[][] valBytes) { + this.keys = keys; + this.valBytes = valBytes; + } + + /** @return {@link Message} wraps array for {@code hist}. */ + static @Nullable DistributedMetaStorageHistoryItemMessage[] toMessages(@Nullable DistributedMetaStorageHistoryItem[] hist) { + if (hist == null) + return null; + + var res = new DistributedMetaStorageHistoryItemMessage[hist.length]; + + for (int i = 0; i < hist.length; ++i) + res[i] = new DistributedMetaStorageHistoryItemMessage(hist[i].keys, hist[i].valBytesArr); + + return res; + } +} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageImpl.java index 990a26d9b2386..575300772a905 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageImpl.java @@ -22,7 +22,9 @@ import java.util.Arrays; import java.util.BitSet; import java.util.Collections; +import java.util.Iterator; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.UUID; import java.util.concurrent.ConcurrentHashMap; @@ -564,25 +566,8 @@ private void onMetaStorageReadyForWrite(ReadWriteMetastorage metastorage) { lock.readLock().lock(); try { - if (isClient) { - Serializable data = new DistributedMetaStorageJoiningNodeData( - getBaselineTopologyId(), - ver, - EMPTY_ARRAY - ); - - dataBag.addJoiningNodeData(COMPONENT_ID, data); - - return; - } - - Serializable data = new DistributedMetaStorageJoiningNodeData( - getBaselineTopologyId(), - ver, - histCache.toArray() - ); - - dataBag.addJoiningNodeData(COMPONENT_ID, data); + dataBag.addJoiningNodeData(COMPONENT_ID, new DistributedMetaStorageJoiningNodeData( + getBaselineTopologyId(), ver, isClient ? EMPTY_ARRAY : histCache.toArray())); } finally { lock.readLock().unlock(); @@ -641,9 +626,9 @@ private int getBaselineTopologyId() { if (!isPersistenceEnabled) return null; - DistributedMetaStorageVersion remoteVer = joiningData.ver; + DistributedMetaStorageVersion remoteVer = new DistributedMetaStorageVersion(joiningData.dVerId, joiningData.dVerHash); - DistributedMetaStorageHistoryItem[] remoteHist = joiningData.hist; + DistributedMetaStorageHistoryItem[] remoteHist = DistributedMetaStorageHistoryItem.fromMessages(joiningData.hist); int remoteHistSize = remoteHist.length; @@ -725,7 +710,7 @@ else if (remoteBltId < locBltId) } if (errorMsg == null) - errorMsg = validatePayload(joiningData); + errorMsg = validatePayload(remoteHist); return (errorMsg == null) ? null : new IgniteNodeValidationResult(node.id(), errorMsg); } @@ -735,11 +720,11 @@ else if (remoteBltId < locBltId) } /** - * @param joiningData Joining data to validate. + * @param remoteHist Joining history data to validate. * @return {@code null} if contained data is valid otherwise error message. */ - private String validatePayload(DistributedMetaStorageJoiningNodeData joiningData) { - for (DistributedMetaStorageHistoryItem item : joiningData.hist) { + private String validatePayload(DistributedMetaStorageHistoryItem[] remoteHist) { + for (DistributedMetaStorageHistoryItem item : remoteHist) { for (int i = 0; i < item.keys().length; i++) { try { unmarshal(marshaller, item.valuesBytesArray()[i]); @@ -766,19 +751,17 @@ private String validatePayload(DistributedMetaStorageJoiningNodeData joiningData DistributedMetaStorageJoiningNodeData joiningData = discoData.joiningNodeData(); - DistributedMetaStorageVersion remoteVer = joiningData.ver; - lock.writeLock().lock(); try { DistributedMetaStorageVersion locVer = ver; - if (remoteVer.id() > locVer.id()) { - DistributedMetaStorageHistoryItem[] hist = joiningData.hist; + if (joiningData.dVerId > locVer.id()) { + DistributedMetaStorageHistoryItem[] hist = DistributedMetaStorageHistoryItem.fromMessages(joiningData.hist); - if (remoteVer.id() - locVer.id() <= hist.length) { - for (long v = locVer.id() + 1; v <= remoteVer.id(); v++) { - int hv = (int)(v - remoteVer.id() + hist.length - 1); + if (joiningData.dVerId - locVer.id() <= hist.length) { + for (long v = locVer.id() + 1; v <= joiningData.dVerId; v++) { + int hv = (int)(v - joiningData.dVerId + hist.length - 1); try { completeWrite(hist[hv]); @@ -788,8 +771,10 @@ private String validatePayload(DistributedMetaStorageJoiningNodeData joiningData } } } - else - assert false : "Joining node is too far ahead [remoteVer=" + remoteVer + "]"; + else { + assert false : "Joining node is too far ahead [remoteVerId=" + joiningData.dVerId + ", remoteVerHash=" + + joiningData.dVerHash + "]"; + } } } finally { @@ -821,30 +806,26 @@ private String validatePayload(DistributedMetaStorageJoiningNodeData joiningData DistributedMetaStorageJoiningNodeData joiningData = discoData.joiningNodeData(); - DistributedMetaStorageVersion remoteVer = joiningData.ver; - lock.readLock().lock(); try { DistributedMetaStorageVersion locVer = ver; - if (remoteVer.id() >= locVer.id()) { - Serializable nodeData = new DistributedMetaStorageClusterNodeData(remoteVer, null, null, null); + if (joiningData.dVerId >= locVer.id()) { + var rmtVer = new DistributedMetaStorageVersion(joiningData.dVerId, joiningData.dVerHash); - dataBag.addGridCommonData(COMPONENT_ID, nodeData); + dataBag.addGridCommonData(COMPONENT_ID, new DistributedMetaStorageClusterNodeData(rmtVer, null, null, null)); } else { - if (locVer.id() - remoteVer.id() <= histCache.size() && !dataBag.isJoiningNodeClient()) { - DistributedMetaStorageHistoryItem[] updates = history(remoteVer.id() + 1, locVer.id()); - - Serializable nodeData = new DistributedMetaStorageClusterNodeData(ver, null, null, updates); + if (locVer.id() - joiningData.dVerId <= histCache.size() && !dataBag.isJoiningNodeClient()) { + DistributedMetaStorageHistoryItem[] updates = history(joiningData.dVerId + 1, locVer.id()); - dataBag.addGridCommonData(COMPONENT_ID, nodeData); + dataBag.addGridCommonData(COMPONENT_ID, new DistributedMetaStorageClusterNodeData(ver, null, null, updates)); } else { DistributedMetaStorageVersion ver0 = ver; - DistributedMetaStorageKeyValuePair[] fullData = bridge.localFullData(); + Map fullData = bridge.localFullData(); DistributedMetaStorageHistoryItem[] hist; @@ -853,7 +834,7 @@ private String validatePayload(DistributedMetaStorageJoiningNodeData joiningData else hist = history(ver.id() - histCache.size() + 1, locVer.id()); - Serializable nodeData = new DistributedMetaStorageClusterNodeData(ver0, fullData, hist, null); + var nodeData = new DistributedMetaStorageClusterNodeData(ver0, fullData, hist, null); dataBag.addGridCommonData(COMPONENT_ID, nodeData); } @@ -943,7 +924,7 @@ private DistributedMetaStorageHistoryItem[] history(long startVer, long actualVe * {@link InMemoryCachedDistributedMetaStorageBridge#localFullData()} invoked on {@link #bridge}. */ @TestOnly - private DistributedMetaStorageKeyValuePair[] localFullData() { + public Map localFullData() { return bridge.localFullData(); } @@ -961,30 +942,34 @@ private DistributedMetaStorageKeyValuePair[] localFullData() { DistributedMetaStorageClusterNodeData nodeData = data.commonData(); if (nodeData != null) { - if (nodeData.fullData != null) { - ver = nodeData.ver; + if (nodeData.fullDataKeys != null) { + assert nodeData.fullDataValsBytes != null && nodeData.fullDataValsBytes.length == nodeData.fullDataKeys.length; + + ver = new DistributedMetaStorageVersion(nodeData.dVerId, nodeData.dVerHash); - notifyListenersBeforeReadyForWrite(nodeData.fullData); + notifyListenersBeforeReadyForWrite(nodeData.fullDataKeys, nodeData.fullDataValsBytes); bridge.writeFullNodeData(nodeData); } + // Cached unwrapped history. + DistributedMetaStorageHistoryItem[] newHist = EMPTY_ARRAY; + if (nodeData.hist != null) { - clearHistoryCache(); + newHist = DistributedMetaStorageHistoryItem.fromMessages(nodeData.hist); - for (int i = 0, len = nodeData.hist.length; i < len; i++) { - DistributedMetaStorageHistoryItem histItem = nodeData.hist[i]; + clearHistoryCache(); - addToHistoryCache(ver.id() + i - (len - 1), histItem); - } + for (int i = 0, len = newHist.length; i < len; i++) + addToHistoryCache(ver.id() + i - (len - 1), newHist[i]); } - if (isPersistenceEnabled && nodeData.fullData != null) - dataWriter.addUpdateTask(nodeData); + if (isPersistenceEnabled && nodeData.fullDataKeys != null) + dataWriter.addUpdateTask(ver, newHist, nodeData.fullDataKeys, nodeData.fullDataValsBytes); if (nodeData.updates != null) { - for (DistributedMetaStorageHistoryItem update : nodeData.updates) - completeWrite(update); + for (DistributedMetaStorageHistoryItem item : DistributedMetaStorageHistoryItem.fromMessages(nodeData.updates)) + completeWrite(item); } } else if (!isClient && ver.id() > 0) { @@ -1178,9 +1163,7 @@ private RuntimeException criticalError(Throwable e) { * @param histItem {@code } pair to process. * @throws IgniteCheckedException In case of IO/unmarshalling errors. */ - private void completeWrite( - DistributedMetaStorageHistoryItem histItem - ) throws IgniteCheckedException { + private void completeWrite(DistributedMetaStorageHistoryItem histItem) throws IgniteCheckedException { assert lock.writeLock().isHeldByCurrentThread(); histItem = optimizeHistoryItem(histItem); @@ -1318,30 +1301,31 @@ void clearHistoryCache() { /** * Notify listeners on node start. Even if there was no data restoring. * - * @param newData Data about which listeners should be notified. + * @param newDataKeys Data keys about which listeners should be notified. + * @param newDataVals Data values about which listeners should be notified. */ - private void notifyListenersBeforeReadyForWrite( - DistributedMetaStorageKeyValuePair[] newData - ) throws IgniteCheckedException { + private void notifyListenersBeforeReadyForWrite(String[] newDataKeys, byte[][] newDataVals) throws IgniteCheckedException { assert lock.isWriteLockedByCurrentThread(); - DistributedMetaStorageKeyValuePair[] oldData = bridge.localFullData(); + Map oldData = bridge.localFullData(); - int oldIdx = 0, newIdx = 0; + Iterator> oldDataIt = oldData.entrySet().iterator(); + Map.Entry oldDataEntry = oldDataIt.hasNext() ? oldDataIt.next() : null; + int newIdx = 0; - while (oldIdx < oldData.length && newIdx < newData.length) { - String oldKey = oldData[oldIdx].key; - byte[] oldValBytes = oldData[oldIdx].valBytes; + while (oldDataEntry != null && newIdx < newDataKeys.length) { + String oldKey = oldDataEntry.getKey(); + byte[] oldValBytes = oldDataEntry.getValue(); - String newKey = newData[newIdx].key; - byte[] newValBytes = newData[newIdx].valBytes; + String newKey = newDataKeys[newIdx]; + byte[] newValBytes = newDataVals[newIdx]; int c = oldKey.compareTo(newKey); if (c < 0) { notifyListeners(oldKey, () -> unmarshal(marshaller, oldValBytes), () -> null); - ++oldIdx; + oldDataEntry = oldDataIt.hasNext() ? oldDataIt.next() : null; } else if (c > 0) { notifyListeners(newKey, () -> null, () -> unmarshal(marshaller, newValBytes)); @@ -1349,25 +1333,26 @@ else if (c > 0) { ++newIdx; } else { - notifyListeners( - oldKey, - () -> unmarshal(marshaller, oldValBytes), - () -> unmarshal(marshaller, newValBytes)); + notifyListeners(oldKey, () -> unmarshal(marshaller, oldValBytes), () -> unmarshal(marshaller, newValBytes)); - ++oldIdx; + oldDataEntry = oldDataIt.hasNext() ? oldDataIt.next() : null; ++newIdx; } } - for (; oldIdx < oldData.length; ++oldIdx) { - byte[] oldValBytes = oldData[oldIdx].valBytes; - notifyListeners(oldData[oldIdx].key, () -> unmarshal(marshaller, oldValBytes), () -> null); + while (oldDataEntry != null) { + byte[] oldDataVal = oldDataEntry.getValue(); + + notifyListeners(oldDataEntry.getKey(), () -> unmarshal(marshaller, oldDataVal), () -> null); + + oldDataEntry = oldDataIt.hasNext() ? oldDataIt.next() : null; } - for (; newIdx < newData.length; ++newIdx) { - byte[] newValBytes = newData[newIdx].valBytes; - notifyListeners(newData[newIdx].key, () -> null, () -> unmarshal(marshaller, newValBytes)); + for (; newIdx < newDataKeys.length; ++newIdx) { + byte[] newDataVal = newDataVals[newIdx]; + + notifyListeners(newDataKeys[newIdx], () -> null, () -> unmarshal(marshaller, newDataVal)); } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageJoiningNodeData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageJoiningNodeData.java index c9be2671d3105..11748fb9c99ae 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageJoiningNodeData.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageJoiningNodeData.java @@ -17,30 +17,45 @@ package org.apache.ignite.internal.processors.metastorage.persistence; -import java.io.Serializable; +import java.io.Externalizable; +import org.apache.ignite.internal.Order; +import org.apache.ignite.internal.processors.cache.persistence.metastorage.MetaStorage; +import org.apache.ignite.internal.util.tostring.GridToStringInclude; +import org.apache.ignite.plugin.extensions.communication.Message; /** - * Distributed metastorage data that joining node sends to cluster. + * Distributed metastorage data message that a joining node sends to a cluster. To reduce the messages number, contains + * plain representation of {@link DistributedMetaStorageVersion}. And {@link Message}s wraps of {@link DistributedMetaStorageHistoryItem}s. + * The original data holders are a {@link Externalizable} and persistent by {@link MetaStorage} with the dedicated + * code-generated serializers. Thus, we do not make them directly a {@link Message}. + * + * @see DmsDataWriter#write(String, byte[]) + * @see MetaStorage#write(String, Serializable) */ -@SuppressWarnings("PublicField") -class DistributedMetaStorageJoiningNodeData implements Serializable { - /** */ - private static final long serialVersionUID = 0L; +public class DistributedMetaStorageJoiningNodeData implements Message { + /** Baseline topology id of node, {@code -1} if baseline topology is null. */ + @Order(0) + int bltId; - /** - * Baseline topology id of node, {@code -1} if baseline topology is null. - */ - public final int bltId; + /** @see DistributedMetaStorageVersion#id */ + @Order(1) + @GridToStringInclude + long dVerId; - /** - * Distributed metastorage version of joining node. - */ - public final DistributedMetaStorageVersion ver; + /** @see DistributedMetaStorageVersion#hash */ + @Order(2) + @GridToStringInclude + long dVerHash; - /** - * Available history of joining node. - */ - public final DistributedMetaStorageHistoryItem[] hist; + /** Available history of joining node. */ + @Order(3) + @GridToStringInclude + DistributedMetaStorageHistoryItemMessage[] hist; + + /** For serialization purposes. */ + public DistributedMetaStorageJoiningNodeData() { + // No-op. + } /** */ public DistributedMetaStorageJoiningNodeData( @@ -48,8 +63,14 @@ public DistributedMetaStorageJoiningNodeData( DistributedMetaStorageVersion ver, DistributedMetaStorageHistoryItem[] hist ) { + assert ver != null; + assert hist != null; + this.bltId = bltId; - this.ver = ver; - this.hist = hist; + + dVerId = ver.id; + dVerHash = ver.hash; + + this.hist = DistributedMetaStorageHistoryItemMessage.toMessages(hist); } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageKeyValuePair.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageKeyValuePair.java deleted file mode 100644 index 9adbe9b9069d0..0000000000000 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageKeyValuePair.java +++ /dev/null @@ -1,70 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one or more - * contributor license agreements. See the NOTICE file distributed with - * this work for additional information regarding copyright ownership. - * The ASF licenses this file to You under the Apache License, Version 2.0 - * (the "License"); you may not use this file except in compliance with - * the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ - -package org.apache.ignite.internal.processors.metastorage.persistence; - -import java.io.Serializable; -import java.util.Arrays; -import org.apache.ignite.internal.util.tostring.GridToStringInclude; -import org.apache.ignite.internal.util.typedef.internal.S; - -/** */ -@SuppressWarnings("PublicField") -final class DistributedMetaStorageKeyValuePair implements Serializable { - /** */ - private static final long serialVersionUID = 0L; - - /** */ - public static final DistributedMetaStorageKeyValuePair[] EMPTY_ARRAY = {}; - - /** */ - @GridToStringInclude - public final String key; - - /** */ - @GridToStringInclude - public final byte[] valBytes; - - /** */ - public DistributedMetaStorageKeyValuePair(String key, byte[] valBytes) { - this.key = key; - this.valBytes = valBytes; - } - - /** {@inheritDoc} */ - @Override public boolean equals(Object o) { - if (this == o) - return true; - - if (o == null || getClass() != o.getClass()) - return false; - - DistributedMetaStorageKeyValuePair pair = (DistributedMetaStorageKeyValuePair)o; - - return key.equals(pair.key) && Arrays.equals(valBytes, pair.valBytes); - } - - /** {@inheritDoc} */ - @Override public int hashCode() { - return 31 * key.hashCode() + Arrays.hashCode(valBytes); - } - - /** {@inheritDoc} */ - @Override public String toString() { - return S.toString(DistributedMetaStorageKeyValuePair.class, this); - } -} diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageVersion.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageVersion.java index 88b4a37084626..31102ba9f6482 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageVersion.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DistributedMetaStorageVersion.java @@ -67,7 +67,7 @@ public DistributedMetaStorageVersion() { * @param id Id. * @param hash Hash. */ - private DistributedMetaStorageVersion(long id, long hash) { + DistributedMetaStorageVersion(long id, long hash) { this.id = id; this.hash = hash; } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DmsDataWriter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DmsDataWriter.java index 06f7337e884ae..5ed14fa54a200 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DmsDataWriter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/DmsDataWriter.java @@ -140,29 +140,29 @@ public void addUpdateTask(DistributedMetaStorageHistoryItem histItem) { } /** */ - public void addUpdateTask(DistributedMetaStorageClusterNodeData fullNodeData) { - assert fullNodeData.fullData != null; - assert fullNodeData.hist != null; - + public void addUpdateTask( + DistributedMetaStorageVersion ver, + DistributedMetaStorageHistoryItem[] hist, + String[] newDataKeys, + byte[][] newDataVals + ) { addToQueue(newDmsTask(() -> { metastorage.writeRaw(cleanupGuardKey(), DUMMY_VALUE); doCleanup(); - for (DistributedMetaStorageKeyValuePair item : fullNodeData.fullData) - metastorage.writeRaw(localKey(item.key), item.valBytes); - - for (int i = 0, len = fullNodeData.hist.length; i < len; i++) { - DistributedMetaStorageHistoryItem histItem = fullNodeData.hist[i]; + for (int i = 0; i < newDataKeys.length; ++i) + metastorage.writeRaw(localKey(newDataKeys[i]), newDataVals[i]); - long histItemVer = fullNodeData.ver.id() + i - (len - 1); + for (int i = 0, len = hist.length; i < len; i++) { + long histItemVer = ver.id() + i - (len - 1); - metastorage.write(historyItemKey(histItemVer), histItem); + metastorage.write(historyItemKey(histItemVer), hist[i]); } - metastorage.write(versionKey(), fullNodeData.ver); + metastorage.write(versionKey(), ver); - workerDmsVer = fullNodeData.ver; + workerDmsVer = ver; metastorage.remove(cleanupGuardKey()); })); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/InMemoryCachedDistributedMetaStorageBridge.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/InMemoryCachedDistributedMetaStorageBridge.java index 298d0b5c50900..0147c410b49cb 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/InMemoryCachedDistributedMetaStorageBridge.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/metastorage/persistence/InMemoryCachedDistributedMetaStorageBridge.java @@ -105,22 +105,22 @@ public void write(String globalKey, @Nullable byte[] valBytes) { * Returns all {@code } pairs currently stored in distributed metastorage. Values are not unmarshalled. * All keys are sorted in ascending order. * - * @return Array of all keys and values. + * @return All the keys and values. */ - public DistributedMetaStorageKeyValuePair[] localFullData() { - return cache.entrySet().stream().map( - entry -> new DistributedMetaStorageKeyValuePair(entry.getKey(), entry.getValue()) - ).toArray(DistributedMetaStorageKeyValuePair[]::new); + public Map localFullData() { + return cache; } /** */ public void writeFullNodeData(DistributedMetaStorageClusterNodeData fullNodeData) { - assert fullNodeData.fullData != null; + assert fullNodeData.fullDataKeys != null; + assert fullNodeData.fullDataValsBytes != null; + assert fullNodeData.fullDataKeys.length == fullNodeData.fullDataValsBytes.length; cache.clear(); - for (DistributedMetaStorageKeyValuePair item : fullNodeData.fullData) - cache.put(item.key, item.valBytes); + for (int i = 0; i < fullNodeData.fullDataKeys.length; ++i) + cache.put(fullNodeData.fullDataKeys[i], fullNodeData.fullDataValsBytes[i]); } /** */ diff --git a/modules/core/src/main/resources/META-INF/classnames.properties b/modules/core/src/main/resources/META-INF/classnames.properties index c49e06ca2f70c..fc1966fef4b4e 100644 --- a/modules/core/src/main/resources/META-INF/classnames.properties +++ b/modules/core/src/main/resources/META-INF/classnames.properties @@ -1608,7 +1608,6 @@ org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaSto org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageClusterNodeData org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageHistoryItem org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageJoiningNodeData -org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageKeyValuePair org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageUpdateAckMessage org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageUpdateMessage org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageVersion diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/metastorage/DistributedMetaStorageTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/metastorage/DistributedMetaStorageTest.java index 49f576ef60322..4a88aede8fc01 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/metastorage/DistributedMetaStorageTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/metastorage/DistributedMetaStorageTest.java @@ -17,9 +17,7 @@ package org.apache.ignite.internal.processors.metastorage; -import java.lang.reflect.Method; -import java.util.Arrays; -import java.util.Comparator; +import java.util.TreeMap; import java.util.UUID; import java.util.concurrent.Callable; import java.util.concurrent.ThreadLocalRandom; @@ -561,9 +559,9 @@ protected DistributedMetaStorage metastorage(int i) { * Assert that two nodes have the same internal state in {@link DistributedMetaStorage}. */ protected void assertDistributedMetastoragesAreEqual(IgniteEx ignite1, IgniteEx ignite2) throws Exception { - DistributedMetaStorage distributedMetastorage1 = ignite1.context().distributedMetastorage(); + DistributedMetaStorageImpl distributedMetastorage1 = (DistributedMetaStorageImpl)ignite1.context().distributedMetastorage(); - DistributedMetaStorage distributedMetastorage2 = ignite2.context().distributedMetastorage(); + DistributedMetaStorageImpl distributedMetastorage2 = (DistributedMetaStorageImpl)ignite2.context().distributedMetastorage(); Object ver1 = U.field(distributedMetastorage1, "ver"); @@ -577,17 +575,15 @@ protected void assertDistributedMetastoragesAreEqual(IgniteEx ignite1, IgniteEx assertEquals(histCache1, histCache2); - Method fullDataMtd = U.findNonPublicMethod(DistributedMetaStorageImpl.class, "localFullData"); + var fullData1 = distributedMetastorage1.localFullData(); - Object[] fullData1 = (Object[])fullDataMtd.invoke(distributedMetastorage1); + var fullData2 = distributedMetastorage2.localFullData(); - Object[] fullData2 = (Object[])fullDataMtd.invoke(distributedMetastorage2); - - assertEqualsCollections(Arrays.asList(fullData1), Arrays.asList(fullData2)); + assertEqualsMaps(fullData1, fullData2); // Also check that arrays are sorted. - Arrays.sort(fullData1, Comparator.comparing(o -> U.field(o, "key"))); + fullData1 = new TreeMap<>(fullData1); - assertEqualsCollections(Arrays.asList(fullData1), Arrays.asList(fullData2)); + assertEqualsMaps(fullData1, fullData2); } } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/metastorage/persistence/DmsDataWriterTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/metastorage/persistence/DmsDataWriterTest.java index 29f36e40589c0..ed48a4e8c2348 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/metastorage/persistence/DmsDataWriterTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/metastorage/persistence/DmsDataWriterTest.java @@ -203,12 +203,7 @@ public void testUpdateFullNodeData() throws Exception { DistributedMetaStorageVersion ver = INITIAL_VERSION.nextVersion(update); - dmsDataWriter.addUpdateTask(new DistributedMetaStorageClusterNodeData( - ver, - new DistributedMetaStorageKeyValuePair[] {toKeyValuePair(update)}, - new DistributedMetaStorageHistoryItem[] {update}, - new DistributedMetaStorageHistoryItem[] {histItem("key4", "val4")} // Has to be ignored. - )); + dmsDataWriter.addUpdateTask(ver, new DistributedMetaStorageHistoryItem[] {update}, update.keys, update.valBytesArr); stopWorker(); @@ -329,13 +324,6 @@ public void testHalt() throws Exception { assertEquals("val1", metastorage.read(localKey("key1"))); } - /** */ - private DistributedMetaStorageKeyValuePair toKeyValuePair(DistributedMetaStorageHistoryItem histItem) { - assertEquals(1, histItem.keys().length); - - return new DistributedMetaStorageKeyValuePair(histItem.keys()[0], histItem.valuesBytesArray()[0]); - } - /** */ private void write(String key, String val) throws IgniteCheckedException { dmsDataWriter.addUpdateTask(histItem(key, val)); diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/metastorage/persistence/InMemoryCachedDistributedMetaStorageBridgeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/metastorage/persistence/InMemoryCachedDistributedMetaStorageBridgeTest.java index 4c5815e1dedb1..e223fae03067a 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/metastorage/persistence/InMemoryCachedDistributedMetaStorageBridgeTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/metastorage/persistence/InMemoryCachedDistributedMetaStorageBridgeTest.java @@ -19,6 +19,7 @@ import java.util.ArrayList; import java.util.List; +import java.util.Map; import org.apache.ignite.IgniteCheckedException; import org.apache.ignite.marshaller.jdk.JdkMarshaller; import org.junit.Before; @@ -33,12 +34,14 @@ import static org.apache.ignite.internal.processors.metastorage.persistence.DistributedMetaStorageVersion.INITIAL_VERSION; import static org.apache.ignite.internal.processors.metastorage.persistence.DmsDataWriter.DUMMY_VALUE; import static org.apache.ignite.testframework.junits.common.GridCommonAbstractTest.TEST_JDK_MARSHALLER; +import static org.apache.ignite.testframework.junits.common.GridCommonAbstractTest.assertEqualsMaps; import static org.hamcrest.CoreMatchers.instanceOf; import static org.hamcrest.CoreMatchers.is; import static org.junit.Assert.assertArrayEquals; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; +import static org.junit.Assert.assertTrue; /** */ public class InMemoryCachedDistributedMetaStorageBridgeTest { @@ -101,13 +104,13 @@ public void testLocalFullData() throws Exception { bridge.write("key2", valBytes2); bridge.write("key1", valBytes1); - DistributedMetaStorageKeyValuePair[] exp = { - new DistributedMetaStorageKeyValuePair("key1", valBytes1), - new DistributedMetaStorageKeyValuePair("key2", valBytes2), - new DistributedMetaStorageKeyValuePair("key3", valBytes3) - }; + Map exp = Map.of( + "key1", valBytes1, + "key2", valBytes2, + "key3", valBytes3 + ); - assertArrayEquals(exp, bridge.localFullData()); + assertEqualsMaps(exp, bridge.localFullData()); } /** */ @@ -117,9 +120,7 @@ public void testWriteFullNodeData() throws Exception { bridge.writeFullNodeData(new DistributedMetaStorageClusterNodeData( DistributedMetaStorageVersion.INITIAL_VERSION, - new DistributedMetaStorageKeyValuePair[] { - new DistributedMetaStorageKeyValuePair("newKey", marshaller.marshal("newVal")) - }, + Map.of("newKey", marshaller.marshal("newVal")), DistributedMetaStorageHistoryItem.EMPTY_ARRAY, null )); @@ -140,7 +141,7 @@ public void testReadInitialDataAfterFailedCleanup() throws Exception { bridge.readInitialData(metastorage); - assertArrayEquals(DistributedMetaStorageKeyValuePair.EMPTY_ARRAY, bridge.localFullData()); + assertTrue(bridge.localFullData().isEmpty()); } /** */ @@ -156,7 +157,7 @@ public void testReadInitialData1() throws Exception { bridge.readInitialData(metastorage); - assertEquals(1, bridge.localFullData().length); + assertEquals(1, bridge.localFullData().size()); assertEquals("val1", bridge.read("key1")); } @@ -175,7 +176,7 @@ public void testReadInitialData2() throws Exception { bridge.readInitialData(metastorage); - assertEquals(1, bridge.localFullData().length); + assertEquals(1, bridge.localFullData().size()); assertEquals("val1", bridge.read("key1")); } @@ -195,7 +196,7 @@ public void testReadInitialData3() throws Exception { bridge.readInitialData(metastorage); - assertEquals(1, bridge.localFullData().length); + assertEquals(1, bridge.localFullData().size()); assertEquals("val1", bridge.read("key1")); } diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java index 6432a528b3918..55e53c7d44cb1 100755 --- a/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java +++ b/modules/core/src/test/java/org/apache/ignite/testframework/junits/common/GridCommonAbstractTest.java @@ -1890,15 +1890,19 @@ protected static void assertEqualsCollectionsIgnoringOrder(Collection exp * @param exp Expected. * @param act Actual. */ - protected static void assertEqualsMaps(Map exp, Map act) { + public static void assertEqualsMaps(Map exp, Map act) { if (exp.size() != act.size()) fail("Maps are not equal:\nExpected:\t" + exp + "\nActual:\t" + act); for (Map.Entry e : exp.entrySet()) { if (!act.containsKey(e.getKey())) fail("Maps are not equal (missing key " + e.getKey() + "):\nExpected:\t" + exp + "\nActual:\t" + act); - else if (!Objects.equals(e.getValue(), act.get(e.getKey()))) - fail("Maps are not equal (key " + e.getKey() + "):\nExpected:\t" + exp + "\nActual:\t" + act); + + assertEqualsArraysAware( + "Maps are not equal (key " + e.getKey() + "):\nExpected:\t" + exp + "\nActual:\t" + act, + e.getValue(), + act.get(e.getKey()) + ); } }