You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by na...@apache.org on 2026/03/20 13:27:23 UTC
(ignite) branch master updated: IGNITE-27631 Use MessageSerializer for SingleNodeMessage (#12827)
This is an automated email from the ASF dual-hosted git repository.
namelchev pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new e149edcf34b IGNITE-27631 Use MessageSerializer for SingleNodeMessage (#12827)
e149edcf34b is described below
commit e149edcf34b01f8efb57575821ae304d0aad6426
Author: Nikita Amelchev <ns...@gmail.com>
AuthorDate: Fri Mar 20 16:27:17 2026 +0300
IGNITE-27631 Use MessageSerializer for SingleNodeMessage (#12827)
---
modules/core/pom.xml | 14 +++
.../communication/GridIoMessageFactory.java | 41 ++++++-
.../discovery/DiscoveryMessageFactory.java | 46 +++++++-
.../managers/encryption/GridEncryptionManager.java | 23 ++--
.../managers/encryption/GroupKeyChangeProcess.java | 18 +--
.../internal/processors/cache/StoredCacheData.java | 2 +
.../snapshot/DataStreamerUpdatesHandler.java | 11 +-
.../snapshot/DataStreamerUpdatesHandlerResult.java | 49 ++++++++
.../snapshot/IgniteSnapshotManager.java | 43 ++-----
.../snapshot/IncrementalSnapshotVerifyResult.java | 53 +++++++--
.../SnapshotCheckHandlersNodeResponse.java | 50 ++++++++
.../snapshot/SnapshotCheckHandlersResponse.java | 51 +++++++++
.../SnapshotCheckPartitionHashesResponse.java | 69 +++++++++++
.../persistence/snapshot/SnapshotCheckProcess.java | 68 ++++-------
.../snapshot/SnapshotCheckResponse.java | 74 ++++++++++++
.../persistence/snapshot/SnapshotChecker.java | 5 +-
.../persistence/snapshot/SnapshotHandler.java | 3 +-
.../snapshot/SnapshotHandlerResult.java | 34 ++++--
.../persistence/snapshot/SnapshotMetadata.java | 2 +
.../snapshot/SnapshotMetadataResponse.java | 67 +++++++++++
.../snapshot/SnapshotOperationResponse.java | 51 +++++++++
.../SnapshotPartitionsQuickVerifyHandler.java | 12 +-
.../snapshot/SnapshotPartitionsVerifyHandler.java | 24 ++--
.../SnapshotPartitionsVerifyHandlerResponse.java | 69 +++++++++++
.../snapshot/SnapshotRestoreOperationResponse.java | 93 +++++++++++++++
.../snapshot/SnapshotRestoreProcess.java | 89 ++++++---------
.../PerformanceStatisticsProcessor.java | 3 +-
.../util/distributed/DistributedProcess.java | 3 +-
.../internal/util/distributed/FullMessage.java | 46 ++++++--
.../util/distributed/SingleNodeMessage.java | 126 +++++----------------
.../direct/DirectMarshallingMessagesTest.java | 108 ------------------
.../snapshot/IgniteClusterSnapshotHandlerTest.java | 66 ++++++-----
.../snapshot/IncrementalSnapshotTest.java | 3 +-
.../DistributedProcessClientAwaitTest.java | 24 ++--
.../DistributedProcessCoordinatorLeftTest.java | 18 +--
.../DistributedProcessErrorHandlingTest.java | 15 ++-
.../util/distributed/MessagesPluginProvider.java | 64 +++++++++++
.../util/distributed/TestIntegerMessage.java | 49 ++++++++
.../internal/util/distributed/TestUuidMessage.java | 50 ++++++++
.../testsuites/IgniteMarshallerSelfTestSuite.java | 2 -
.../ignite/testsuites/IgniteUtilSelfTestSuite.java | 6 +-
41 files changed, 1155 insertions(+), 489 deletions(-)
diff --git a/modules/core/pom.xml b/modules/core/pom.xml
index 5ebf5ee4628..421d3071a87 100644
--- a/modules/core/pom.xml
+++ b/modules/core/pom.xml
@@ -35,6 +35,7 @@
<properties>
<ignite.update.notifier.product>apache-ignite</ignite.update.notifier.product>
<ignite.generated.source.path>${project.build.directory}/generated-sources/codegen</ignite.generated.source.path>
+ <ignite.generated.test.source.path>${project.build.directory}/generated-test-sources/codegen</ignite.generated.test.source.path>
</properties>
<dependencies>
@@ -376,6 +377,18 @@
</sources>
</configuration>
</execution>
+ <execution>
+ <id>add-generated-test-sources</id>
+ <phase>generate-test-sources</phase>
+ <goals>
+ <goal>add-test-source</goal>
+ </goals>
+ <configuration>
+ <sources>
+ <source>${ignite.generated.test.source.path}</source>
+ </sources>
+ </configuration>
+ </execution>
</executions>
</plugin>
@@ -383,6 +396,7 @@
<artifactId>maven-compiler-plugin</artifactId>
<configuration>
<generatedSourcesDirectory>${ignite.generated.source.path}</generatedSourcesDirectory>
+ <generatedTestSourcesDirectory>${ignite.generated.test.source.path}</generatedTestSourcesDirectory>
<annotationProcessors>
<annotationProcessor>org.apache.ignite.internal.MessageProcessor</annotationProcessor>
<annotationProcessor>org.apache.ignite.internal.idto.IgniteDataTransferObjectProcessor</annotationProcessor>
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
index 65480e0b48e..52e124e95c4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java
@@ -231,12 +231,34 @@ import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPr
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequestSerializer;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponse;
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponseSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.DataStreamerUpdatesHandlerResult;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.DataStreamerUpdatesHandlerResultSerializer;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotAwareMessage;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotAwareMessageSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotVerifyResult;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotVerifyResultMarshallableSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersNodeResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersNodeResponseSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersResponseSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckPartitionHashesResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckPartitionHashesResponseMarshallableSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckResponseSerializer;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFilesFailureMessage;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFilesFailureMessageSerializer;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFilesRequestMessage;
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFilesRequestMessageSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandlerResult;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandlerResultSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataResponseMarshallableSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperationResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperationResponseSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyHandlerResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyHandlerResponseMarshallableSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreOperationResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreOperationResponseMarshallableSerializer;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryRequest;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryRequestSerializer;
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryResponse;
@@ -332,6 +354,7 @@ import org.apache.ignite.internal.util.GridPartitionStateMapSerializer;
import org.apache.ignite.internal.util.UUIDCollectionMessage;
import org.apache.ignite.internal.util.UUIDCollectionMessageSerializer;
import org.apache.ignite.internal.util.distributed.SingleNodeMessage;
+import org.apache.ignite.internal.util.distributed.SingleNodeMessageSerializer;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
@@ -522,7 +545,7 @@ public class GridIoMessageFactory implements MessageFactoryProvider {
factory.register(GridIoSecurityAwareMessage.TYPE_CODE, GridIoSecurityAwareMessage::new,
new GridIoSecurityAwareMessageMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
factory.register(SessionChannelMessage.TYPE_CODE, SessionChannelMessage::new, new SessionChannelMessageSerializer());
- factory.register(SingleNodeMessage.TYPE_CODE, SingleNodeMessage::new);
+ factory.register(SingleNodeMessage.TYPE_CODE, SingleNodeMessage::new, new SingleNodeMessageSerializer());
factory.register((short)177, TcpInverseConnectionResponseMessage::new, new TcpInverseConnectionResponseMessageSerializer());
factory.register(SnapshotFilesRequestMessage.TYPE_CODE, SnapshotFilesRequestMessage::new,
new SnapshotFilesRequestMessageSerializer());
@@ -566,6 +589,22 @@ public class GridIoMessageFactory implements MessageFactoryProvider {
factory.register(GridPartitionStateMap.TYPE_CODE, GridPartitionStateMap::new, new GridPartitionStateMapSerializer());
factory.register(GridDhtPartitionMap.TYPE_CODE, GridDhtPartitionMap::new, new GridDhtPartitionMapSerializer());
factory.register(GridDhtPartitionFullMap.TYPE_CODE, GridDhtPartitionFullMap::new, new GridDhtPartitionFullMapSerializer());
+ factory.register((short)520, SnapshotOperationResponse::new, new SnapshotOperationResponseSerializer());
+ factory.register((short)521, SnapshotHandlerResult::new, new SnapshotHandlerResultSerializer());
+ factory.register((short)522, DataStreamerUpdatesHandlerResult::new, new DataStreamerUpdatesHandlerResultSerializer());
+ factory.register((short)523, SnapshotCheckResponse::new, new SnapshotCheckResponseSerializer());
+ factory.register((short)524, IncrementalSnapshotVerifyResult::new,
+ new IncrementalSnapshotVerifyResultMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
+ factory.register((short)525, SnapshotRestoreOperationResponse::new,
+ new SnapshotRestoreOperationResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
+ factory.register((short)526, SnapshotMetadataResponse::new,
+ new SnapshotMetadataResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
+ factory.register((short)527, SnapshotCheckPartitionHashesResponse::new,
+ new SnapshotCheckPartitionHashesResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
+ factory.register((short)528, SnapshotCheckHandlersResponse::new, new SnapshotCheckHandlersResponseSerializer());
+ factory.register((short)529, SnapshotCheckHandlersNodeResponse::new, new SnapshotCheckHandlersNodeResponseSerializer());
+ factory.register((short)530, SnapshotPartitionsVerifyHandlerResponse::new,
+ new SnapshotPartitionsVerifyHandlerResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
// [-3..119] [124..129] [-23..-28] [-36..-55] [183..188] - this
// [120..123] - DR
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryMessageFactory.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryMessageFactory.java
index 26f744bac52..f4038f87f86 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryMessageFactory.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/discovery/DiscoveryMessageFactory.java
@@ -45,6 +45,30 @@ import org.apache.ignite.internal.processors.cache.binary.MetadataRemoveProposed
import org.apache.ignite.internal.processors.cache.binary.MetadataRemoveProposedMessageSerializer;
import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateAcceptedMessage;
import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateAcceptedMessageSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.DataStreamerUpdatesHandlerResult;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.DataStreamerUpdatesHandlerResultSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotVerifyResult;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotVerifyResultMarshallableSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersNodeResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersNodeResponseSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersResponseSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckPartitionHashesResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckPartitionHashesResponseMarshallableSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckResponseSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandlerResult;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandlerResultSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataResponseMarshallableSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperationResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperationResponseSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyHandlerResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyHandlerResponseMarshallableSerializer;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreOperationResponse;
+import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreOperationResponseMarshallableSerializer;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersionSerializer;
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessage;
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessageSerializer;
import org.apache.ignite.internal.processors.continuous.StopRoutineAckDiscoveryMessage;
@@ -81,6 +105,8 @@ import org.apache.ignite.internal.processors.query.schema.operation.SchemaIndexC
import org.apache.ignite.internal.processors.query.schema.operation.SchemaIndexCreateOperationMarshallableSerializer;
import org.apache.ignite.internal.processors.query.schema.operation.SchemaIndexDropOperation;
import org.apache.ignite.internal.processors.query.schema.operation.SchemaIndexDropOperationSerializer;
+import org.apache.ignite.internal.util.distributed.FullMessage;
+import org.apache.ignite.internal.util.distributed.FullMessageSerializer;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
@@ -236,6 +262,9 @@ public class DiscoveryMessageFactory implements MessageFactoryProvider {
factory.register((short)28, TcpDiscoveryClientReconnectMessage::new, new TcpDiscoveryClientReconnectMessageSerializer());
factory.register((short)29, TcpDiscoveryNodeAddedMessage::new,
new TcpDiscoveryNodeAddedMessageMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
+ factory.register((short)30, FullMessage::new, new FullMessageSerializer());
+
+ factory.register((short)86, GridCacheVersion::new, new GridCacheVersionSerializer());
// DiscoveryCustomMessage
factory.register((short)500, CacheStatisticsModeChangeMessage::new, new CacheStatisticsModeChangeMessageSerializer());
@@ -260,6 +289,21 @@ public class DiscoveryMessageFactory implements MessageFactoryProvider {
factory.register((short)517, MappingAcceptedMessage::new, new MappingAcceptedMessageSerializer());
factory.register((short)518, MappingProposedMessage::new, new MappingProposedMessageSerializer());
factory.register((short)519, MarshallerMappingItem::new, new MarshallerMappingItemSerializer());
-
+ factory.register((short)520, SnapshotOperationResponse::new, new SnapshotOperationResponseSerializer());
+ factory.register((short)521, SnapshotHandlerResult::new, new SnapshotHandlerResultSerializer());
+ factory.register((short)522, DataStreamerUpdatesHandlerResult::new, new DataStreamerUpdatesHandlerResultSerializer());
+ factory.register((short)523, SnapshotCheckResponse::new, new SnapshotCheckResponseSerializer());
+ factory.register((short)524, IncrementalSnapshotVerifyResult::new,
+ new IncrementalSnapshotVerifyResultMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
+ factory.register((short)525, SnapshotRestoreOperationResponse::new,
+ new SnapshotRestoreOperationResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
+ factory.register((short)526, SnapshotMetadataResponse::new,
+ new SnapshotMetadataResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
+ factory.register((short)527, SnapshotCheckPartitionHashesResponse::new,
+ new SnapshotCheckPartitionHashesResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
+ factory.register((short)528, SnapshotCheckHandlersResponse::new, new SnapshotCheckHandlersResponseSerializer());
+ factory.register((short)529, SnapshotCheckHandlersNodeResponse::new, new SnapshotCheckHandlersNodeResponseSerializer());
+ factory.register((short)530, SnapshotPartitionsVerifyHandlerResponse::new,
+ new SnapshotPartitionsVerifyHandlerResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java
index 738b3c738cb..c5a217d3725 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java
@@ -71,6 +71,7 @@ import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteFutureCancelledException;
import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.spi.IgniteNodeValidationResult;
import org.apache.ignite.spi.IgniteSpiException;
import org.apache.ignite.spi.discovery.DiscoveryDataBag;
@@ -203,10 +204,10 @@ public class GridEncryptionManager extends GridManagerAdapter<EncryptionSpi> imp
* Master key change prepare process. Checks that all server nodes have the same new master key and then starts
* finish process.
*/
- private DistributedProcess<MasterKeyChangeRequest, EmptyResult> prepareMKChangeProc;
+ private DistributedProcess<MasterKeyChangeRequest, Message> prepareMKChangeProc;
/** Process to perform the master key change. Changes master key and reencrypt group keys. */
- private DistributedProcess<MasterKeyChangeRequest, EmptyResult> performMKChangeProc;
+ private DistributedProcess<MasterKeyChangeRequest, Message> performMKChangeProc;
/**
* A two-phase distributed process that rotates the encryption keys of specified cache groups and initiates
@@ -1488,7 +1489,7 @@ public class GridEncryptionManager extends GridManagerAdapter<EncryptionSpi> imp
* @param req Request.
* @return Result future.
*/
- private IgniteInternalFuture<EmptyResult> prepareMasterKeyChange(MasterKeyChangeRequest req) {
+ private IgniteInternalFuture<Message> prepareMasterKeyChange(MasterKeyChangeRequest req) {
if (masterKeyChangeRequest != null) {
return new GridFinishedFuture<>(new IgniteException("Master key change was rejected. " +
"The previous change was not completed."));
@@ -1524,7 +1525,7 @@ public class GridEncryptionManager extends GridManagerAdapter<EncryptionSpi> imp
ctx.localNodeId() + ']', e));
}
- return new GridFinishedFuture<>(new EmptyResult());
+ return new GridFinishedFuture<>();
}
/**
@@ -1534,7 +1535,7 @@ public class GridEncryptionManager extends GridManagerAdapter<EncryptionSpi> imp
* @param res Results.
* @param err Errors.
*/
- private void finishPrepareMasterKeyChange(UUID id, Map<UUID, EmptyResult> res, Map<UUID, Throwable> err) {
+ private void finishPrepareMasterKeyChange(UUID id, Map<UUID, Message> res, Map<UUID, Throwable> err) {
if (!err.isEmpty()) {
if (masterKeyChangeRequest != null && masterKeyChangeRequest.requestId().equals(id))
masterKeyChangeRequest = null;
@@ -1551,7 +1552,7 @@ public class GridEncryptionManager extends GridManagerAdapter<EncryptionSpi> imp
* @param req Request.
* @return Result future.
*/
- private IgniteInternalFuture<EmptyResult> performMasterKeyChange(MasterKeyChangeRequest req) {
+ private IgniteInternalFuture<Message> performMasterKeyChange(MasterKeyChangeRequest req) {
if (masterKeyChangeRequest == null || !masterKeyChangeRequest.equals(req))
return new GridFinishedFuture<>(new IgniteException("Unknown master key change was rejected."));
@@ -1569,7 +1570,7 @@ public class GridEncryptionManager extends GridManagerAdapter<EncryptionSpi> imp
masterKeyDigest = req.digest();
- return new GridFinishedFuture<>(new EmptyResult());
+ return new GridFinishedFuture<>();
}
/**
@@ -1579,7 +1580,7 @@ public class GridEncryptionManager extends GridManagerAdapter<EncryptionSpi> imp
* @param res Results.
* @param err Errors.
*/
- private void finishPerformMasterKeyChange(UUID id, Map<UUID, EmptyResult> res, Map<UUID, Throwable> err) {
+ private void finishPerformMasterKeyChange(UUID id, Map<UUID, Message> res, Map<UUID, Throwable> err) {
completeMasterKeyChangeFuture(id, err);
}
@@ -1818,12 +1819,6 @@ public class GridEncryptionManager extends GridManagerAdapter<EncryptionSpi> imp
}
}
- /** */
- protected static class EmptyResult implements Serializable {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
- }
-
/** */
protected static class NodeEncryptionKeys implements Serializable {
/** */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GroupKeyChangeProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GroupKeyChangeProcess.java
index 415216d631a..ac1fb0250d3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GroupKeyChangeProcess.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GroupKeyChangeProcess.java
@@ -25,7 +25,6 @@ import java.util.UUID;
import org.apache.ignite.IgniteException;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.managers.encryption.GridEncryptionManager.EmptyResult;
import org.apache.ignite.internal.managers.encryption.GridEncryptionManager.KeyChangeFuture;
import org.apache.ignite.internal.processors.cache.CacheGroupDescriptor;
import org.apache.ignite.internal.processors.cache.DynamicCacheDescriptor;
@@ -41,6 +40,7 @@ import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteFutureCancelledException;
+import org.apache.ignite.plugin.extensions.communication.Message;
import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CACHE_GROUP_KEY_CHANGE_FINISH;
import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CACHE_GROUP_KEY_CHANGE_PREPARE;
@@ -54,10 +54,10 @@ class GroupKeyChangeProcess {
private final GridKernalContext ctx;
/** Cache group encyption key change prepare phase. */
- private final DistributedProcess<ChangeCacheEncryptionRequest, EmptyResult> prepareGKChangeProc;
+ private final DistributedProcess<ChangeCacheEncryptionRequest, Message> prepareGKChangeProc;
/** Cache group encyption key change perform phase. */
- private final DistributedProcess<ChangeCacheEncryptionRequest, EmptyResult> performGKChangeProc;
+ private final DistributedProcess<ChangeCacheEncryptionRequest, Message> performGKChangeProc;
/** Group encryption keys. */
private final CacheGroupEncryptionKeys keys;
@@ -183,7 +183,7 @@ class GroupKeyChangeProcess {
* @param req Request.
* @return Result future.
*/
- private IgniteInternalFuture<EmptyResult> prepare(ChangeCacheEncryptionRequest req) {
+ private IgniteInternalFuture<Message> prepare(ChangeCacheEncryptionRequest req) {
if (ctx.clientNode())
return new GridFinishedFuture<>();
@@ -251,7 +251,7 @@ class GroupKeyChangeProcess {
ctx.encryption().addGroupKey(req.groupIds()[i], grpKey);
}
- return new GridFinishedFuture<>(new EmptyResult());
+ return new GridFinishedFuture<>();
});
}
@@ -268,7 +268,7 @@ class GroupKeyChangeProcess {
* @param res Results.
* @param err Errors.
*/
- private void finishPrepare(UUID id, Map<UUID, EmptyResult> res, Map<UUID, Throwable> err) {
+ private void finishPrepare(UUID id, Map<UUID, Message> res, Map<UUID, Throwable> err) {
if (!err.isEmpty()) {
if (req != null && req.requestId().equals(id))
req = null;
@@ -285,7 +285,7 @@ class GroupKeyChangeProcess {
* @param req Request.
* @return Result future.
*/
- private IgniteInternalFuture<EmptyResult> perform(ChangeCacheEncryptionRequest req) {
+ private IgniteInternalFuture<Message> perform(ChangeCacheEncryptionRequest req) {
if (this.req == null || !this.req.equals(req))
return new GridFinishedFuture<>(new IgniteException("Unknown cache group key change was rejected."));
@@ -303,7 +303,7 @@ class GroupKeyChangeProcess {
this.req = null;
}
- return new GridFinishedFuture<>(new EmptyResult());
+ return new GridFinishedFuture<>();
}
/**
@@ -313,7 +313,7 @@ class GroupKeyChangeProcess {
* @param res Results.
* @param err Errors.
*/
- private void finishPerform(UUID id, Map<UUID, EmptyResult> res, Map<UUID, Throwable> err) {
+ private void finishPerform(UUID id, Map<UUID, Message> res, Map<UUID, Throwable> err) {
completeFuture(id, err, fut);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/StoredCacheData.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/StoredCacheData.java
index 5f53e7385ea..9a9e85e6d5e 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/StoredCacheData.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/StoredCacheData.java
@@ -36,6 +36,8 @@ import org.apache.ignite.marshaller.jdk.JdkMarshaller;
* {@link CacheConfiguration} and additional information about cache which is not a part of configuration.
* This class is {@link Serializable} and is intended to be read-written with {@link JdkMarshaller}
* in order to be serialization wise agnostic to further additions or removals of fields.
+ * <p>
+ * All changes must be made with the respect of RU rules.
*/
public class StoredCacheData implements Serializable, CdcCacheEvent {
/** */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/DataStreamerUpdatesHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/DataStreamerUpdatesHandler.java
index 8bbb9ff642d..f4bb0171689 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/DataStreamerUpdatesHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/DataStreamerUpdatesHandler.java
@@ -24,14 +24,13 @@ import java.util.UUID;
import java.util.stream.Collectors;
import org.apache.ignite.internal.util.typedef.F;
-import static java.lang.Boolean.TRUE;
import static org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandlerType.CREATE;
/**
* A snapshot haldler that monitors and warns of inconsistent by nature updates from DataStreamer which can issue
* data inconsistency in snapshot.
*/
-public class DataStreamerUpdatesHandler implements SnapshotHandler<Boolean> {
+public class DataStreamerUpdatesHandler implements SnapshotHandler<DataStreamerUpdatesHandlerResult> {
/** */
public static final String WRN_MSG = "DataStreamer with property 'allowOverwrite' set to `false` was working " +
"during the snapshot creation. Such streaming updates are inconsistent by nature and should be successfully " +
@@ -44,14 +43,14 @@ public class DataStreamerUpdatesHandler implements SnapshotHandler<Boolean> {
}
/** {@inheritDoc} */
- @Override public Boolean invoke(SnapshotHandlerContext ctx) {
- return ctx.streamerWarning();
+ @Override public DataStreamerUpdatesHandlerResult invoke(SnapshotHandlerContext ctx) {
+ return new DataStreamerUpdatesHandlerResult(ctx.streamerWarning());
}
/** {@inheritDoc} */
- @Override public void complete(String name, Map<UUID, SnapshotHandlerResult<Boolean>> results)
+ @Override public void complete(String name, Map<UUID, SnapshotHandlerResult<DataStreamerUpdatesHandlerResult>> results)
throws SnapshotWarningException {
- Collection<UUID> nodes = F.viewReadOnly(results.entrySet(), Entry::getKey, e -> TRUE.equals(e.getValue().data()));
+ Collection<UUID> nodes = F.viewReadOnly(results.entrySet(), Entry::getKey, e -> e.getValue().data().streamerWarning());
if (!F.isEmpty(nodes)) {
throw new SnapshotWarningException(WRN_MSG + " Updates from DataStreamer detected on the nodes: " +
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/DataStreamerUpdatesHandlerResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/DataStreamerUpdatesHandlerResult.java
new file mode 100644
index 00000000000..06e2f90ac72
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/DataStreamerUpdatesHandlerResult.java
@@ -0,0 +1,49 @@
+/*
+ * 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.cache.persistence.snapshot;
+
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
+import org.apache.ignite.plugin.extensions.communication.Message;
+
+/** */
+public class DataStreamerUpdatesHandlerResult implements Message {
+ /** */
+ @Order(0)
+ boolean streamerWarning;
+
+ /** Default constructor for {@link GridIoMessageFactory}. */
+ public DataStreamerUpdatesHandlerResult() {
+ // No-op.
+ }
+
+ /** @param streamerWarning Streamer warning flag. */
+ public DataStreamerUpdatesHandlerResult(boolean streamerWarning) {
+ this.streamerWarning = streamerWarning;
+ }
+
+ /** @return Streamer warning flag. */
+ public boolean streamerWarning() {
+ return streamerWarning;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 522;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
index ed01716a600..7445cb07582 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteSnapshotManager.java
@@ -181,6 +181,7 @@ import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.marshaller.Marshaller;
import org.apache.ignite.metric.MetricRegistry;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.resources.IgniteInstanceResource;
import org.apache.ignite.spi.encryption.EncryptionSpi;
import org.apache.ignite.spi.systemview.view.SnapshotView;
@@ -2874,7 +2875,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
/** Snapshot operation handlers. */
protected static class SnapshotHandlers {
/** Snapshot operation handlers. */
- private final Map<SnapshotHandlerType, List<SnapshotHandler<Object>>> handlers = new EnumMap<>(SnapshotHandlerType.class);
+ private final Map<SnapshotHandlerType, List<SnapshotHandler<?>>> handlers = new EnumMap<>(SnapshotHandlerType.class);
/** Executor service used to invoke handlers in parallel. */
private ExecutorService execSvc;
@@ -2896,12 +2897,12 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
registerHandler(new SnapshotPartitionsQuickVerifyHandler(ctx.cache().context()));
// Register custom handlers.
- SnapshotHandler<Object>[] extHnds = (SnapshotHandler<Object>[])ctx.plugins().extensions(SnapshotHandler.class);
+ SnapshotHandler<Message>[] extHnds = (SnapshotHandler<Message>[])ctx.plugins().extensions(SnapshotHandler.class);
if (extHnds == null)
return;
- for (SnapshotHandler<Object> extHnd : extHnds)
+ for (SnapshotHandler<Message> extHnd : extHnds)
registerHandler(extHnd);
}
@@ -2911,17 +2912,17 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
* @return Results from all handlers with the specified type.
* @throws IgniteCheckedException if parallel execution was failed.
*/
- protected @Nullable Map<String, SnapshotHandlerResult<Object>> invokeAll(
+ protected @Nullable Map<String, SnapshotHandlerResult<Message>> invokeAll(
SnapshotHandlerType type,
SnapshotHandlerContext ctx
) throws IgniteCheckedException {
- List<SnapshotHandler<Object>> handlers = this.handlers.get(type);
+ List<SnapshotHandler<?>> handlers = this.handlers.get(type);
if (F.isEmpty(handlers))
return null;
if (handlers.size() == 1) {
- SnapshotHandler<Object> hnd = handlers.get(0);
+ SnapshotHandler<?> hnd = handlers.get(0);
return F.asMap(hnd.getClass().getName(), invoke(hnd, ctx));
}
@@ -2952,7 +2953,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
if (res.isEmpty())
return;
- List<SnapshotHandler<Object>> hnds = handlers.get(type);
+ List<SnapshotHandler<?>> hnds = handlers.get(type);
if (hnds == null || hnds.size() != res.size()) {
throw new IgniteCheckedException("Snapshot handlers configuration mismatch (number of local snapshot " +
@@ -2995,7 +2996,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
* @param hnd Snapshot operation handler.
* @param ctx Snapshot operation handler context.
*/
- private SnapshotHandlerResult<Object> invoke(SnapshotHandler<Object> hnd, SnapshotHandlerContext ctx) {
+ private SnapshotHandlerResult<Message> invoke(SnapshotHandler<?> hnd, SnapshotHandlerContext ctx) {
try {
return new SnapshotHandlerResult<>(hnd.invoke(ctx), null);
}
@@ -3007,7 +3008,7 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
}
/** */
- private void registerHandler(SnapshotHandler hnd) {
+ private void registerHandler(SnapshotHandler<?> hnd) {
handlers.computeIfAbsent(hnd.type(), v -> new ArrayList<>()).add(hnd);
}
}
@@ -4122,30 +4123,6 @@ public class IgniteSnapshotManager extends GridCacheSharedManagerAdapter
}
}
- /** */
- private static class SnapshotOperationResponse implements Serializable {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
-
- /** Results of single-node handlers execution. */
- private final Map<String, SnapshotHandlerResult<Object>> hndResults;
-
- /** Default constructor. */
- public SnapshotOperationResponse() {
- this(null);
- }
-
- /** @param hndResults Results of single-node handlers execution. */
- public SnapshotOperationResponse(Map<String, SnapshotHandlerResult<Object>> hndResults) {
- this.hndResults = hndResults;
- }
-
- /** @return Results of single-node handlers execution. */
- public @Nullable Map<String, SnapshotHandlerResult<Object>> handlerResults() {
- return hndResults;
- }
- }
-
/** Snapshot operation start message. */
private static class SnapshotStartDiscoveryMessage extends InitMessage<SnapshotOperationRequest>
implements SnapshotDiscoveryMessage {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerifyResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerifyResult.java
index 7a3db0d2485..c08861590e4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerifyResult.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotVerifyResult.java
@@ -17,34 +17,48 @@
package org.apache.ignite.internal.processors.cache.persistence.snapshot;
-import java.io.Serializable;
import java.util.Collection;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.communication.ErrorMessage;
import org.apache.ignite.internal.pagemem.wal.record.DataEntry;
import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
import org.apache.ignite.internal.processors.cache.verify.TransactionsHashRecord;
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
/** */
-public class IncrementalSnapshotVerifyResult implements Serializable {
- /** */
- private static final long serialVersionUID = 0L;
-
+public class IncrementalSnapshotVerifyResult implements MarshallableMessage {
/** Transaction hashes collection. */
private Collection<TransactionsHashRecord> txHashRes;
+ /** */
+ @Order(0)
+ byte[] txHashResBytes;
+
/**
* Partition hashes collection. Value is a hash of data entries {@link DataEntry} from WAL segments included
* into the incremental snapshot.
*/
private Collection<PartitionHashRecord> partHashRes;
+ /** */
+ @Order(1)
+ byte[] partHashResBytes;
+
/** Partially committed transactions' collection. */
- private Collection<GridCacheVersion> partiallyCommittedTxs;
+ @Order(2)
+ Collection<GridCacheVersion> partiallyCommittedTxs;
/** Occurred exceptions. */
- private Collection<Exception> exceptions;
+ @Order(3)
+ Collection<ErrorMessage> exceptions;
- /** */
+ /** Default constructor for {@link MessageFactory}. */
public IncrementalSnapshotVerifyResult() {
// No-op.
}
@@ -59,7 +73,7 @@ public class IncrementalSnapshotVerifyResult implements Serializable {
this.txHashRes = txHashRes;
this.partHashRes = partHashRes;
this.partiallyCommittedTxs = partiallyCommittedTxs;
- this.exceptions = exceptions;
+ this.exceptions = exceptions == null ? null : F.viewReadOnly(exceptions, ErrorMessage::new);
}
/** */
@@ -77,8 +91,23 @@ public class IncrementalSnapshotVerifyResult implements Serializable {
return partiallyCommittedTxs;
}
- /** */
- public Collection<Exception> exceptions() {
- return exceptions;
+ /** {@inheritDoc} */
+ @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+ txHashResBytes = U.marshal(marsh, txHashRes);
+ partHashResBytes = U.marshal(marsh, partHashRes);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+ if (txHashResBytes != null)
+ txHashRes = U.unmarshal(marsh, txHashResBytes, clsLdr);
+
+ if (partHashResBytes != null)
+ partHashRes = U.unmarshal(marsh, partHashResBytes, clsLdr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 524;
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckHandlersNodeResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckHandlersNodeResponse.java
new file mode 100644
index 00000000000..eb9e2a585f3
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckHandlersNodeResponse.java
@@ -0,0 +1,50 @@
+/*
+ * 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.cache.persistence.snapshot;
+
+import java.util.Map;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+
+/** */
+public class SnapshotCheckHandlersNodeResponse implements Message {
+ /** Check result per handler name. */
+ @Order(0)
+ Map<String, SnapshotHandlerResult<Message>> hndRes;
+
+ /** Default constructor for {@link MessageFactory}. */
+ public SnapshotCheckHandlersNodeResponse() {
+ // No-op.
+ }
+
+ /** */
+ public SnapshotCheckHandlersNodeResponse(Map<String, SnapshotHandlerResult<Message>> hndRes) {
+ this.hndRes = hndRes;
+ }
+
+ /** */
+ public Map<String, SnapshotHandlerResult<Message>> handlerResults() {
+ return hndRes;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 529;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckHandlersResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckHandlersResponse.java
new file mode 100644
index 00000000000..e35436c5b22
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckHandlersResponse.java
@@ -0,0 +1,51 @@
+/*
+ * 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.cache.persistence.snapshot;
+
+import java.util.Map;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+
+/** */
+public class SnapshotCheckHandlersResponse implements Message {
+ /** Per metas result: snapshot part's consistent id -> check result per handler name. */
+ @Order(0)
+ Map<String, SnapshotCheckHandlersNodeResponse> perMetaResults;
+
+ /** Default constructor for {@link MessageFactory}. */
+ public SnapshotCheckHandlersResponse() {
+ // No-op.
+ }
+
+ /** */
+ public SnapshotCheckHandlersResponse(Map<String, Map<String, SnapshotHandlerResult<Message>>> res) {
+ perMetaResults = F.viewReadOnly(res, SnapshotCheckHandlersNodeResponse::new);
+ }
+
+ /** */
+ public Map<String, Map<String, SnapshotHandlerResult<Message>>> handlerResults() {
+ return F.viewReadOnly(perMetaResults, SnapshotCheckHandlersNodeResponse::handlerResults);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 528;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckPartitionHashesResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckPartitionHashesResponse.java
new file mode 100644
index 00000000000..c38ac9e2f9f
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckPartitionHashesResponse.java
@@ -0,0 +1,69 @@
+/*
+ * 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.cache.persistence.snapshot;
+
+import java.util.Map;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.management.cache.PartitionKey;
+import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+
+/** */
+public class SnapshotCheckPartitionHashesResponse implements MarshallableMessage {
+ /** Per metas result: consistent id -> check results per partition key. */
+ private Map<String, Map<PartitionKey, PartitionHashRecord>> perMetaResults;
+
+ /** */
+ @Order(0)
+ byte[] perMetaResultsBytes;
+
+ /** Default constructor for {@link MessageFactory}. */
+ public SnapshotCheckPartitionHashesResponse() {
+ // No-op.
+ }
+
+ /** */
+ public SnapshotCheckPartitionHashesResponse(Map<String, Map<PartitionKey, PartitionHashRecord>> results) {
+ perMetaResults = results;
+ }
+
+ /** */
+ public Map<String, Map<PartitionKey, PartitionHashRecord>> perMetaResults() {
+ return perMetaResults;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+ perMetaResultsBytes = U.marshal(marsh, perMetaResults);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+ if (perMetaResultsBytes != null)
+ perMetaResults = U.unmarshal(marsh, perMetaResultsBytes, clsLdr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 527;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java
index 1baacc7d978..ab3e1a96029 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckProcess.java
@@ -17,7 +17,6 @@
package org.apache.ignite.internal.processors.cache.persistence.snapshot;
-import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
@@ -55,6 +54,7 @@ import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.CU;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.CHECK_SNAPSHOT_METAS;
@@ -185,7 +185,7 @@ public class SnapshotCheckProcess {
for (Map.Entry<UUID, SnapshotCheckResponse> resEntry : results.entrySet()) {
UUID nodeId = resEntry.getKey();
- SnapshotCheckResponse incResp = resEntry.getValue();
+ SnapshotCheckResponse<IncrementalSnapshotVerifyResult> incResp = resEntry.getValue();
if (incResp == null || !requiredNodes.contains(nodeId))
continue;
@@ -218,7 +218,7 @@ public class SnapshotCheckProcess {
Map<UUID, Map<Object, Map<String, SnapshotHandlerResult<?>>>> reduced = new HashMap<>(results.size(), 1.0f);
for (Map.Entry<UUID, SnapshotCheckResponse> respEntry : results.entrySet()) {
- SnapshotCheckResponse nodeResp = respEntry.getValue();
+ SnapshotCheckResponse<SnapshotCheckHandlersResponse> nodeResp = respEntry.getValue();
if (nodeResp == null)
continue;
@@ -228,7 +228,7 @@ public class SnapshotCheckProcess {
UUID nodeId = respEntry.getKey();
- Map<String, Map<String, SnapshotHandlerResult<Object>>> cstHndRes = nodeResp.result();
+ Map<String, Map<String, SnapshotHandlerResult<Message>>> cstHndRes = nodeResp.result().handlerResults();
cstHndRes.forEach((consId, respPerConsIdMap) -> {
// Reduced map of the handlers results per snapshot part's consistent id for certain node.
@@ -288,7 +288,7 @@ public class SnapshotCheckProcess {
bldr.exceptions(errors0);
for (Map.Entry<UUID, SnapshotCheckResponse> respEntry : results.entrySet()) {
- SnapshotCheckResponse resp = respEntry.getValue();
+ SnapshotCheckResponse<SnapshotCheckPartitionHashesResponse> resp = respEntry.getValue();
if (resp == null)
continue;
@@ -299,7 +299,7 @@ public class SnapshotCheckProcess {
bldr.addException(node, asException(F.firstValue(resp.exceptions())));
}
- Map<String, Map<PartitionKey, PartitionHashRecord>> partsHashesRes = resp.result();
+ Map<String, Map<PartitionKey, PartitionHashRecord>> partsHashesRes = resp.result().perMetaResults();
partsHashesRes.forEach((consId, partsPerConsId) -> bldr.addPartitionHashes(partsPerConsId));
}
@@ -384,7 +384,7 @@ public class SnapshotCheckProcess {
for (SnapshotMetadata meta : ctx.metas) {
// Run asynchronously to calculate the metric 'total partitions' faster.
kctx.pools().getSnapshotExecutorService().submit(() -> {
- CompletableFuture<Map<PartitionKey, PartitionHashRecord>> metaFut = snpChecker.checkPartitions(
+ CompletableFuture<SnapshotPartitionsVerifyHandlerResponse> metaFut = snpChecker.checkPartitions(
meta,
ctx.locFileTree.get(meta.consistentId()),
ctx.req.groups(),
@@ -399,11 +399,13 @@ public class SnapshotCheckProcess {
if (err != null)
exceptions.put(meta.consistentId(), err);
- else if (!F.isEmpty(res))
- perMetaResults.put(meta.consistentId(), res);
+ else if (!F.isEmpty(res.response()))
+ perMetaResults.put(meta.consistentId(), res.response());
- if (metasProcessed.decrementAndGet() == 0)
- composedFut.complete(new SnapshotCheckResponse(perMetaResults, exceptions));
+ if (metasProcessed.decrementAndGet() == 0) {
+ composedFut.complete(new SnapshotCheckResponse(
+ new SnapshotCheckPartitionHashesResponse(perMetaResults), exceptions));
+ }
});
});
}
@@ -417,7 +419,7 @@ public class SnapshotCheckProcess {
*/
private CompletableFuture<SnapshotCheckResponse> allHandlersFuture(SnapshotCheckContext ctx) {
// Per metas result: snapshot part's consistent id -> check result per handler name.
- Map<String, Map<String, SnapshotHandlerResult<Object>>> perMetaResults = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f);
+ Map<String, Map<String, SnapshotHandlerResult<Message>>> perMetaResults = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f);
// Per consistent id.
Map<String, Throwable> exceptions = new ConcurrentHashMap<>(ctx.metas.size(), 1.0f);
CompletableFuture<SnapshotCheckResponse> composedFut = new CompletableFuture<>();
@@ -426,7 +428,7 @@ public class SnapshotCheckProcess {
for (SnapshotMetadata meta : ctx.metas) {
// Run asynchronously to calculate the metric 'total partitions' faster.
kctx.pools().getSnapshotExecutorService().submit(() -> {
- CompletableFuture<Map<String, SnapshotHandlerResult<Object>>> metaFut = snpChecker.invokeCustomHandlers(
+ CompletableFuture<Map<String, SnapshotHandlerResult<Message>>> metaFut = snpChecker.invokeCustomHandlers(
meta,
ctx.locFileTree.get(meta.consistentId()),
ctx.req.groups(),
@@ -444,7 +446,7 @@ public class SnapshotCheckProcess {
perMetaResults.put(meta.consistentId(), res);
if (metasProcessed.decrementAndGet() == 0)
- composedFut.complete(new SnapshotCheckResponse(perMetaResults, exceptions));
+ composedFut.complete(new SnapshotCheckResponse(new SnapshotCheckHandlersResponse(perMetaResults), exceptions));
});
});
}
@@ -511,7 +513,7 @@ public class SnapshotCheckProcess {
if (err != null)
phaseFut.onDone(err);
else
- phaseFut.onDone(new SnapshotCheckResponse(locMetas, null));
+ phaseFut.onDone(new SnapshotCheckResponse(new SnapshotMetadataResponse(locMetas), null));
});
return phaseFut;
@@ -548,10 +550,11 @@ public class SnapshotCheckProcess {
results.forEach((nodeId, nodeRes) -> {
// A node might be not required. It gives null result. But a required node might have invalid empty result
// which must be validated.
- if (ctx.req.nodes().contains(nodeId) && baseline(nodeId) && !F.isEmpty((Collection<?>)nodeRes.result())) {
+ if (ctx.req.nodes().contains(nodeId) && baseline(nodeId)
+ && !F.isEmpty(((SnapshotMetadataResponse)nodeRes.result()).metadata())) {
assert nodeRes != null;
- metas.put(kctx.cluster().get().node(nodeId), nodeRes.result());
+ metas.put(kctx.cluster().get().node(nodeId), ((SnapshotMetadataResponse)nodeRes.result()).metadata());
}
});
@@ -798,35 +801,4 @@ public class SnapshotCheckProcess {
return S.toString(SnapshotCheckContext.class, this);
}
}
-
- /** A DTO to transfer node's results for the both phases. */
- private static final class SnapshotCheckResponse implements Serializable {
- /** */
- private static final long serialVersionUID = 0L;
-
- /** The result. Is usually a collection or a map of hashes, metas, etc. */
- private final Object result;
-
- /** Exceptions per consistent id. */
- @Nullable private final Map<String, Throwable> exceptions;
-
- /** */
- private SnapshotCheckResponse(Object result, @Nullable Map<String, Throwable> exceptions) {
- assert result instanceof Serializable : "Snapshot check result is not serializable.";
- assert exceptions == null || exceptions instanceof Serializable : "Snapshot check exceptions aren't serializable.";
-
- this.result = result;
- this.exceptions = exceptions == null ? null : Collections.unmodifiableMap(exceptions);
- }
-
- /** @return Exceptions per snapshot part's consistent id. */
- private @Nullable Map<String, Throwable> exceptions() {
- return exceptions;
- }
-
- /** @return Certain phase's and process' result. */
- private <T> T result() {
- return (T)result;
- }
- }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckResponse.java
new file mode 100644
index 00000000000..ca70ba2e13b
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotCheckResponse.java
@@ -0,0 +1,74 @@
+/*
+ * 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.cache.persistence.snapshot;
+
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.Map;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.communication.ErrorMessage;
+import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
+import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.jetbrains.annotations.Nullable;
+
+/** A DTO to transfer node's results for the both phases. */
+public final class SnapshotCheckResponse<T extends Message> implements Message {
+ /** The result. Is usually a collection or a map of hashes, metas, etc. */
+ @Order(0)
+ Message result;
+
+ /** Exceptions per consistent id. */
+ @Order(1)
+ @Nullable Map<String, ErrorMessage> errors;
+
+ /** Default constructor for {@link GridIoMessageFactory}. */
+ public SnapshotCheckResponse() {
+ // No-op.
+ }
+
+ /** */
+ public SnapshotCheckResponse(T result, @Nullable Map<String, Throwable> exceptions) {
+ assert exceptions == null || exceptions instanceof Serializable : "Snapshot check exceptions aren't serializable.";
+
+ this.result = result;
+
+ if (exceptions != null) {
+ HashMap<String, ErrorMessage> errMsgs = new HashMap<>();
+
+ exceptions.forEach((id, e) -> errMsgs.put(id, new ErrorMessage(e)));
+
+ this.errors = errMsgs;
+ }
+ }
+
+ /** @return Exceptions per snapshot part's consistent id. */
+ public @Nullable Map<String, Throwable> exceptions() {
+ return F.viewReadOnly(errors, errMsg -> ErrorMessage.error(errMsg));
+ }
+
+ /** @return Certain phase's and process' result. */
+ public T result() {
+ return (T)result;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 523;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotChecker.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotChecker.java
index 4a3d7bfe7da..4acb019b7cd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotChecker.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotChecker.java
@@ -39,6 +39,7 @@ import org.apache.ignite.internal.processors.cache.persistence.filename.Snapshot
import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
import org.apache.ignite.internal.processors.cache.verify.TransactionsHashRecord;
import org.apache.ignite.internal.util.typedef.F;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.Nullable;
/** */
@@ -168,7 +169,7 @@ public class SnapshotChecker {
*
* @see IgniteSnapshotManager#handlers()
*/
- public CompletableFuture<Map<String, SnapshotHandlerResult<Object>>> invokeCustomHandlers(
+ public CompletableFuture<Map<String, SnapshotHandlerResult<Message>>> invokeCustomHandlers(
SnapshotMetadata meta,
SnapshotFileTree sft,
@Nullable Collection<String> grps,
@@ -201,7 +202,7 @@ public class SnapshotChecker {
}
/** Launches local partitions checking. */
- public CompletableFuture<Map<PartitionKey, PartitionHashRecord>> checkPartitions(
+ public CompletableFuture<SnapshotPartitionsVerifyHandlerResponse> checkPartitions(
SnapshotMetadata meta,
SnapshotFileTree sft,
@Nullable Collection<String> grps,
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotHandler.java
index 4147404de6a..c9e71bc7642 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotHandler.java
@@ -21,6 +21,7 @@ import java.util.Map;
import java.util.UUID;
import org.apache.ignite.IgniteCheckedException;
import org.apache.ignite.plugin.Extension;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.Nullable;
/**
@@ -37,7 +38,7 @@ import org.jetbrains.annotations.Nullable;
*
* @param <T> Type of the local processing result.
*/
-public interface SnapshotHandler<T> extends Extension {
+public interface SnapshotHandler<T extends Message> extends Extension {
/** Snapshot handler type. */
public SnapshotHandlerType type();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotHandlerResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotHandlerResult.java
index 06d38877aa5..aa7f93edf2b 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotHandlerResult.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotHandlerResult.java
@@ -17,7 +17,10 @@
package org.apache.ignite.internal.processors.cache.persistence.snapshot;
-import java.io.Serializable;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.communication.ErrorMessage;
+import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.Nullable;
/**
@@ -26,15 +29,19 @@ import org.jetbrains.annotations.Nullable;
*
* @param <T> Type of the local processing result.
*/
-public class SnapshotHandlerResult<T> implements Serializable {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
-
+public class SnapshotHandlerResult<T extends Message> implements Message {
/** Result of local processing. */
- private final T data;
+ @Order(0)
+ Message data;
/** Processing error. */
- private final Exception err;
+ @Order(1)
+ ErrorMessage errMsg;
+
+ /** Default constructor for {@link GridIoMessageFactory}. */
+ public SnapshotHandlerResult() {
+ // No-op.
+ }
/**
* @param data Result of local processing.
@@ -42,16 +49,23 @@ public class SnapshotHandlerResult<T> implements Serializable {
*/
public SnapshotHandlerResult(@Nullable T data, @Nullable Exception err) {
this.data = data;
- this.err = err;
+
+ if (err != null)
+ errMsg = new ErrorMessage(err);
}
/** @return Result of local processing. */
public @Nullable T data() {
- return data;
+ return (T)data;
}
/** @return Processing error. */
public @Nullable Exception error() {
- return err;
+ return (Exception)ErrorMessage.error(errMsg);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 521;
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadata.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadata.java
index b4691d59941..b74383e4606 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadata.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadata.java
@@ -41,6 +41,8 @@ import org.jetbrains.annotations.Nullable;
/**
* Snapshot metadata file.
+ * <p>
+ * All changes must be made with the respect of RU rules.
*/
public class SnapshotMetadata implements Serializable {
/** Serial version uid. */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataResponse.java
new file mode 100644
index 00000000000..7f396929873
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotMetadataResponse.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.cache.persistence.snapshot;
+
+import java.util.List;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+
+/** */
+public class SnapshotMetadataResponse implements MarshallableMessage {
+ /** */
+ @Order(0)
+ byte[] metadataBytes;
+
+ /** */
+ private List<SnapshotMetadata> metadata;
+
+ /** Default constructor for {@link MessageFactory}. */
+ public SnapshotMetadataResponse() {
+ // No-op.
+ }
+
+ /** */
+ public SnapshotMetadataResponse(List<SnapshotMetadata> metadata) {
+ this.metadata = metadata;
+ }
+
+ /** */
+ public List<SnapshotMetadata> metadata() {
+ return metadata;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+ metadataBytes = U.marshal(marsh, metadata);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+ if (metadataBytes != null)
+ metadata = U.unmarshal(marsh, metadataBytes, clsLdr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 526;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotOperationResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotOperationResponse.java
new file mode 100644
index 00000000000..f595169d4ae
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotOperationResponse.java
@@ -0,0 +1,51 @@
+/*
+ * 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.cache.persistence.snapshot;
+
+import java.util.Map;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.jetbrains.annotations.Nullable;
+
+/** */
+public class SnapshotOperationResponse implements Message {
+ /** Results of single-node handlers execution. */
+ @Order(0)
+ Map<String, SnapshotHandlerResult<Message>> hndResults;
+
+ /** Default constructor for {@link GridIoMessageFactory}. */
+ public SnapshotOperationResponse() {
+ // No-op.
+ }
+
+ /** @param hndResults Results of single-node handlers execution. */
+ public SnapshotOperationResponse(Map<String, SnapshotHandlerResult<Message>> hndResults) {
+ this.hndResults = hndResults;
+ }
+
+ /** @return Results of single-node handlers execution. */
+ public @Nullable Map<String, SnapshotHandlerResult<Message>> handlerResults() {
+ return hndResults;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 520;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsQuickVerifyHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsQuickVerifyHandler.java
index 2a276afec26..bb95f2cef32 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsQuickVerifyHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsQuickVerifyHandler.java
@@ -53,15 +53,15 @@ public class SnapshotPartitionsQuickVerifyHandler extends SnapshotPartitionsVeri
}
/** {@inheritDoc} */
- @Override public Map<PartitionKey, PartitionHashRecord> invoke(SnapshotHandlerContext opCtx)
+ @Override public SnapshotPartitionsVerifyHandlerResponse invoke(SnapshotHandlerContext opCtx)
throws IgniteCheckedException {
// Return null not to check partitions at all if the streamer warning is detected.
if (opCtx.streamerWarning())
return null;
- Map<PartitionKey, PartitionHashRecord> res = super.invoke(opCtx);
+ SnapshotPartitionsVerifyHandlerResponse res = super.invoke(opCtx);
- assert res != null;
+ assert res.response() != null;
return res;
}
@@ -69,7 +69,7 @@ public class SnapshotPartitionsQuickVerifyHandler extends SnapshotPartitionsVeri
/** {@inheritDoc} */
@Override public void complete(
String name,
- Map<UUID, SnapshotHandlerResult<Map<PartitionKey, PartitionHashRecord>>> results
+ Map<UUID, SnapshotHandlerResult<SnapshotPartitionsVerifyHandlerResponse>> results
) throws IgniteCheckedException {
Exception err = results.values().stream().map(SnapshotHandlerResult::error).filter(Objects::nonNull).findAny().orElse(null);
@@ -83,8 +83,8 @@ public class SnapshotPartitionsQuickVerifyHandler extends SnapshotPartitionsVeri
Set<Integer> wrnGrps = new HashSet<>();
Map<PartitionKey, PartitionHashRecord> total = new HashMap<>();
- for (SnapshotHandlerResult<Map<PartitionKey, PartitionHashRecord>> result : results.values()) {
- result.data().forEach((part, val) -> {
+ for (SnapshotHandlerResult<SnapshotPartitionsVerifyHandlerResponse> result : results.values()) {
+ result.data().response().forEach((part, val) -> {
PartitionHashRecord other = total.putIfAbsent(part, val);
if ((other != null && !wrnGrps.contains(part.groupId()))
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandler.java
index 597e95cc58c..84c79b88cc6 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandler.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandler.java
@@ -88,7 +88,7 @@ import static org.apache.ignite.internal.processors.cache.verify.IdleVerifyUtili
/**
* Default snapshot restore handler for checking snapshot partitions consistency.
*/
-public class SnapshotPartitionsVerifyHandler implements SnapshotHandler<Map<PartitionKey, PartitionHashRecord>> {
+public class SnapshotPartitionsVerifyHandler implements SnapshotHandler<SnapshotPartitionsVerifyHandlerResponse> {
/** Shared context. */
protected final GridCacheSharedContext<?, ?> cctx;
@@ -108,7 +108,7 @@ public class SnapshotPartitionsVerifyHandler implements SnapshotHandler<Map<Part
}
/** {@inheritDoc} */
- @Override public Map<PartitionKey, PartitionHashRecord> invoke(SnapshotHandlerContext opCtx) throws IgniteCheckedException {
+ @Override public SnapshotPartitionsVerifyHandlerResponse invoke(SnapshotHandlerContext opCtx) throws IgniteCheckedException {
if (!opCtx.snapshotFileTree().root().exists())
throw new IgniteCheckedException("Snapshot directory doesn't exists: " + opCtx.snapshotFileTree().root());
@@ -185,13 +185,15 @@ public class SnapshotPartitionsVerifyHandler implements SnapshotHandler<Map<Part
if (log.isInfoEnabled())
log.info("Snapshot data integrity check skipped [snpName=" + meta.snapshotName() + ']');
- return Collections.emptyMap();
+ return new SnapshotPartitionsVerifyHandlerResponse(Collections.emptyMap());
}
- return meta.dump()
- ? checkDumpFiles(opCtx, partFiles)
- : checkSnapshotFiles(opCtx.snapshotFileTree(), grpDirs, meta, partFiles, isPunchHoleEnabled(opCtx, grpDirs.keySet()),
- opCtx.progressConsumer() == null ? null : partId -> opCtx.progressConsumer().accept(getClass(), partId));
+ return new SnapshotPartitionsVerifyHandlerResponse(
+ meta.dump()
+ ? checkDumpFiles(opCtx, partFiles)
+ : checkSnapshotFiles(opCtx.snapshotFileTree(), grpDirs, meta, partFiles, isPunchHoleEnabled(opCtx, grpDirs.keySet()),
+ opCtx.progressConsumer() == null ? null : partId -> opCtx.progressConsumer().accept(getClass(), partId))
+ );
}
/** */
@@ -457,13 +459,13 @@ public class SnapshotPartitionsVerifyHandler implements SnapshotHandler<Map<Part
/** {@inheritDoc} */
@Override public void complete(
String name,
- Map<UUID, SnapshotHandlerResult<Map<PartitionKey, PartitionHashRecord>>> results
+ Map<UUID, SnapshotHandlerResult<SnapshotPartitionsVerifyHandlerResponse>> results
) throws IgniteCheckedException {
IdleVerifyResult.Builder bldr = IdleVerifyResult.builder();
- for (Map.Entry<UUID, SnapshotHandlerResult<Map<PartitionKey, PartitionHashRecord>>> e : results.entrySet()) {
+ for (Map.Entry<UUID, SnapshotHandlerResult<SnapshotPartitionsVerifyHandlerResponse>> e : results.entrySet()) {
UUID nodeId = e.getKey();
- SnapshotHandlerResult<Map<PartitionKey, PartitionHashRecord>> res = e.getValue();
+ SnapshotHandlerResult<SnapshotPartitionsVerifyHandlerResponse> res = e.getValue();
if (res.error() != null) {
bldr.addException(cctx.discovery().historicalNode(nodeId), res.error());
@@ -471,7 +473,7 @@ public class SnapshotPartitionsVerifyHandler implements SnapshotHandler<Map<Part
continue;
}
- Map<PartitionKey, PartitionHashRecord> data = res.data();
+ Map<PartitionKey, PartitionHashRecord> data = res.data().response();
bldr.addPartitionHashes(data);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandlerResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandlerResponse.java
new file mode 100644
index 00000000000..d46f63496f3
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotPartitionsVerifyHandlerResponse.java
@@ -0,0 +1,69 @@
+/*
+ * 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.cache.persistence.snapshot;
+
+import java.util.Map;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.management.cache.PartitionKey;
+import org.apache.ignite.internal.processors.cache.verify.PartitionHashRecord;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+
+/** */
+public class SnapshotPartitionsVerifyHandlerResponse implements MarshallableMessage {
+ /** */
+ private Map<PartitionKey, PartitionHashRecord> res;
+
+ /** */
+ @Order(0)
+ byte[] resBytes;
+
+ /** Default constructor for {@link MessageFactory}. */
+ public SnapshotPartitionsVerifyHandlerResponse() {
+ // No-op.
+ }
+
+ /** */
+ public SnapshotPartitionsVerifyHandlerResponse(Map<PartitionKey, PartitionHashRecord> res) {
+ this.res = res;
+ }
+
+ /** */
+ public Map<PartitionKey, PartitionHashRecord> response() {
+ return res;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+ resBytes = U.marshal(marsh, res);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+ if (resBytes != null)
+ res = U.unmarshal(marsh, resBytes, clsLdr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 530;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreOperationResponse.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreOperationResponse.java
new file mode 100644
index 00000000000..f273c90fcb5
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreOperationResponse.java
@@ -0,0 +1,93 @@
+/*
+ * 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.cache.persistence.snapshot;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.processors.cache.StoredCacheData;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.marshaller.Marshaller;
+import org.apache.ignite.plugin.extensions.communication.MarshallableMessage;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+
+/** Snapshot operation prepare response. */
+public class SnapshotRestoreOperationResponse implements MarshallableMessage {
+ /** Cache configurations on local node. */
+ private List<StoredCacheData> ccfgs;
+
+ /** */
+ @Order(0)
+ byte[] ccfgsBytes;
+
+ /** Snapshot metadata files on local node. */
+ private List<SnapshotMetadata> metas;
+
+ /** */
+ @Order(1)
+ byte[] metasBytes;
+
+ /** Default constructor for {@link MessageFactory}. */
+ public SnapshotRestoreOperationResponse() {
+ // No-op.
+ }
+
+ /**
+ * @param ccfgs Cache configurations on local node.
+ * @param metas Snapshot metadata files on local node.
+ */
+ public SnapshotRestoreOperationResponse(
+ Collection<StoredCacheData> ccfgs,
+ Collection<SnapshotMetadata> metas
+ ) {
+ this.ccfgs = new ArrayList<>(ccfgs);
+ this.metas = new ArrayList<>(metas);
+ }
+
+ /** */
+ public List<StoredCacheData> cacheConfigurations() {
+ return ccfgs;
+ }
+
+ /** */
+ public List<SnapshotMetadata> metadata() {
+ return metas;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void prepareMarshal(Marshaller marsh) throws IgniteCheckedException {
+ ccfgsBytes = U.marshal(marsh, ccfgs);
+ metasBytes = U.marshal(marsh, metas);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void finishUnmarshal(Marshaller marsh, ClassLoader clsLdr) throws IgniteCheckedException {
+ if (ccfgsBytes != null)
+ ccfgs = U.unmarshal(marsh, ccfgsBytes, clsLdr);
+
+ if (metasBytes != null)
+ metas = U.unmarshal(marsh, metasBytes, clsLdr);
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 525;
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java
index 1686f3d7028..84d12d3b24c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/SnapshotRestoreProcess.java
@@ -19,7 +19,6 @@ package org.apache.ignite.internal.processors.cache.persistence.snapshot;
import java.io.File;
import java.io.IOException;
-import java.io.Serializable;
import java.nio.ByteBuffer;
import java.nio.file.Files;
import java.nio.file.Path;
@@ -97,6 +96,7 @@ import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteUuid;
import org.apache.ignite.metric.MetricRegistry;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.Nullable;
import static java.util.Optional.ofNullable;
@@ -137,19 +137,19 @@ public class SnapshotRestoreProcess {
private final DistributedProcess<SnapshotOperationRequest, SnapshotRestoreOperationResponse> prepareRestoreProc;
/** Cache group restore preload partitions phase. */
- private final DistributedProcess<UUID, Boolean> preloadProc;
+ private final DistributedProcess<UUID, Message> preloadProc;
/** Cache group restore cache start phase. */
- private final DistributedProcess<UUID, Boolean> cacheStartProc;
+ private final DistributedProcess<UUID, Message> cacheStartProc;
/** Cache group restore cache stop phase. */
- private final DistributedProcess<UUID, Boolean> cacheStopProc;
+ private final DistributedProcess<UUID, Message> cacheStopProc;
/** Incremental snapshot restore phase. */
- private final DistributedProcess<UUID, Boolean> incSnpRestoreProc;
+ private final DistributedProcess<UUID, Message> incSnpRestoreProc;
/** Cache group restore rollback phase. */
- private final DistributedProcess<UUID, Boolean> rollbackRestoreProc;
+ private final DistributedProcess<UUID, Message> rollbackRestoreProc;
/** Logger. */
private final IgniteLogger log;
@@ -841,21 +841,21 @@ public class SnapshotRestoreProcess {
Map<Integer, StoredCacheData> globalCfgs = new HashMap<>();
for (Map.Entry<UUID, SnapshotRestoreOperationResponse> e : res.entrySet()) {
- if (e.getValue().ccfgs != null) {
- for (StoredCacheData cacheData : e.getValue().ccfgs) {
+ if (e.getValue().cacheConfigurations() != null) {
+ for (StoredCacheData cacheData : e.getValue().cacheConfigurations()) {
globalCfgs.put(CU.cacheId(cacheData.config().getName()), cacheData);
opCtx0.dirs.put(CU.cacheGroupId(cacheData.config()), Arrays.asList(ft.cacheStorages(cacheData.config())));
}
}
- if (!F.isEmpty(e.getValue().metas)) {
- e.getValue().metas.stream().filter(SnapshotMetadata::hasCompressedGroups)
+ if (!F.isEmpty(e.getValue().metadata())) {
+ e.getValue().metadata().stream().filter(SnapshotMetadata::hasCompressedGroups)
.forEach(meta -> meta.cacheGroupIds().stream().filter(meta::isGroupWithCompression)
.forEach(opCtx0::addCompressedGroup));
}
- opCtx0.metasPerNode.put(e.getKey(), new ArrayList<>(e.getValue().metas));
+ opCtx0.metasPerNode.put(e.getKey(), new ArrayList<>(e.getValue().metadata()));
}
opCtx0.cfgs = globalCfgs;
@@ -912,12 +912,12 @@ public class SnapshotRestoreProcess {
* @param reqId Request id.
* @return Future which will be completed when the preload ends.
*/
- private IgniteInternalFuture<Boolean> preload(UUID reqId) {
+ private IgniteInternalFuture<Message> preload(UUID reqId) {
if (ctx.clientNode())
return new GridFinishedFuture<>();
SnapshotRestoreContext opCtx0 = opCtx;
- GridFutureAdapter<Boolean> retFut = new GridFutureAdapter<>();
+ GridFutureAdapter<Message> retFut = new GridFutureAdapter<>();
if (opCtx0 == null)
return new GridFinishedFuture<>(new IgniteCheckedException("Snapshot restore process has incorrect restore state: " + reqId));
@@ -1187,7 +1187,7 @@ public class SnapshotRestoreProcess {
Throwable t0 = ofNullable(opCtx0.err.get()).orElse(t);
if (t0 == null)
- retFut.onDone(true);
+ retFut.onDone();
else {
log.error("Unable to restore cache group(s) from a snapshot " +
"[reqId=" + opCtx.reqId + ", snapshot=" + opCtx.snpName + ']', t0);
@@ -1235,7 +1235,7 @@ public class SnapshotRestoreProcess {
* @param res Results.
* @param errs Errors.
*/
- private void finishPreload(UUID reqId, Map<UUID, Boolean> res, Map<UUID, Throwable> errs) {
+ private void finishPreload(UUID reqId, Map<UUID, Message> res, Map<UUID, Throwable> errs) {
if (ctx.clientNode())
return;
@@ -1261,7 +1261,7 @@ public class SnapshotRestoreProcess {
* @param reqId Request ID.
* @return Result future.
*/
- private IgniteInternalFuture<Boolean> cacheStart(UUID reqId) {
+ private IgniteInternalFuture<Message> cacheStart(UUID reqId) {
if (ctx.clientNode())
return new GridFinishedFuture<>();
@@ -1285,8 +1285,13 @@ public class SnapshotRestoreProcess {
// We set the topology node IDs required to successfully start the cache, if any of the required nodes leave
// the cluster during the cache startup, the whole procedure will be rolled back.
- return ctx.cache().dynamicStartCachesByStoredConf(ccfgs, true, true, false,
- IgniteUuid.fromUuid(reqId));
+ return ctx.cache().dynamicStartCachesByStoredConf(ccfgs, true, true, false, IgniteUuid.fromUuid(reqId))
+ .chain(fut -> {
+ if (fut.error() != null)
+ throw F.wrap(fut.error());
+ else
+ return null;
+ });
}
/**
@@ -1294,7 +1299,7 @@ public class SnapshotRestoreProcess {
* @param res Results.
* @param errs Errors.
*/
- private void finishCacheStart(UUID reqId, Map<UUID, Boolean> res, Map<UUID, Throwable> errs) {
+ private void finishCacheStart(UUID reqId, Map<UUID, Message> res, Map<UUID, Throwable> errs) {
if (ctx.clientNode())
return;
@@ -1326,7 +1331,7 @@ public class SnapshotRestoreProcess {
* @param reqId Request ID.
* @return Result future.
*/
- private IgniteInternalFuture<Boolean> cacheStop(UUID reqId) {
+ private IgniteInternalFuture<Message> cacheStop(UUID reqId) {
if (!U.isLocalNodeCoordinator(ctx.discovery()))
return new GridFinishedFuture<>();
@@ -1348,7 +1353,7 @@ public class SnapshotRestoreProcess {
if (fut.error() != null)
throw F.wrap(fut.error());
else
- return true;
+ return null;
});
}
@@ -1357,7 +1362,7 @@ public class SnapshotRestoreProcess {
* @param res Results.
* @param errs Errors.
*/
- private void finishCacheStop(UUID reqId, Map<UUID, Boolean> res, Map<UUID, Throwable> errs) {
+ private void finishCacheStop(UUID reqId, Map<UUID, Message> res, Map<UUID, Throwable> errs) {
if (ctx.clientNode())
return;
@@ -1377,7 +1382,7 @@ public class SnapshotRestoreProcess {
* @param reqId Request ID.
* @return Result future.
*/
- private IgniteInternalFuture<Boolean> incrementalSnapshotRestore(UUID reqId) {
+ private IgniteInternalFuture<Message> incrementalSnapshotRestore(UUID reqId) {
SnapshotRestoreContext opCtx0 = opCtx;
if (ctx.clientNode() || opCtx0 == null || !opCtx0.nodes().contains(ctx.localNodeId()))
@@ -1389,7 +1394,7 @@ public class SnapshotRestoreProcess {
", caches=" + F.viewReadOnly(opCtx0.cfgs, c -> c.config().getName()) + ']');
}
- GridFutureAdapter<Boolean> res = new GridFutureAdapter<>();
+ GridFutureAdapter<Message> res = new GridFutureAdapter<>();
ctx.pools().getSnapshotExecutorService().submit(() -> {
try {
@@ -1406,7 +1411,7 @@ public class SnapshotRestoreProcess {
if (fut.error() != null)
res.onDone(fut.error());
else
- res.onDone(true);
+ res.onDone();
});
if (cp == null)
@@ -1543,7 +1548,7 @@ public class SnapshotRestoreProcess {
* @param res Results.
* @param errs Errors.
*/
- private void finishIncrementalSnapshotRestore(UUID reqId, Map<UUID, Boolean> res, Map<UUID, Throwable> errs) {
+ private void finishIncrementalSnapshotRestore(UUID reqId, Map<UUID, Message> res, Map<UUID, Throwable> errs) {
if (ctx.clientNode())
return;
@@ -1645,7 +1650,7 @@ public class SnapshotRestoreProcess {
* @param reqId Request ID.
* @return Result future.
*/
- private IgniteInternalFuture<Boolean> rollback(UUID reqId) {
+ private IgniteInternalFuture<Message> rollback(UUID reqId) {
if (ctx.clientNode())
return new GridFinishedFuture<>();
@@ -1654,7 +1659,7 @@ public class SnapshotRestoreProcess {
if (opCtx0 == null || F.isEmpty(opCtx0.dirs))
return new GridFinishedFuture<>();
- GridFutureAdapter<Boolean> retFut = new GridFutureAdapter<>();
+ GridFutureAdapter<Message> retFut = new GridFutureAdapter<>();
synchronized (this) {
opCtx0.stopFut = new IgniteFutureImpl<>(retFut.chain(() -> null));
@@ -1692,7 +1697,7 @@ public class SnapshotRestoreProcess {
if (ex != null)
retFut.onDone(ex);
else
- retFut.onDone(true);
+ retFut.onDone();
});
}
catch (RejectedExecutionException e) {
@@ -1710,7 +1715,7 @@ public class SnapshotRestoreProcess {
* @param res Results.
* @param errs Errors.
*/
- private void finishRollback(UUID reqId, Map<UUID, Boolean> res, Map<UUID, Throwable> errs) {
+ private void finishRollback(UUID reqId, Map<UUID, Message> res, Map<UUID, Throwable> errs) {
if (ctx.clientNode())
return;
@@ -1962,30 +1967,6 @@ public class SnapshotRestoreProcess {
}
}
- /** Snapshot operation prepare response. */
- private static class SnapshotRestoreOperationResponse implements Serializable {
- /** Serial version uid. */
- private static final long serialVersionUID = 0L;
-
- /** Cache configurations on local node. */
- private final List<StoredCacheData> ccfgs;
-
- /** Snapshot metadata files on local node. */
- private final List<SnapshotMetadata> metas;
-
- /**
- * @param ccfgs Cache configurations on local node.
- * @param metas Snapshot metadata files on local node.
- */
- public SnapshotRestoreOperationResponse(
- Collection<StoredCacheData> ccfgs,
- Collection<SnapshotMetadata> metas
- ) {
- this.ccfgs = new ArrayList<>(ccfgs);
- this.metas = new ArrayList<>(metas);
- }
- }
-
/** Future will be completed when partition processing ends. */
private static class PartitionRestoreFuture extends CompletableFuture<Path> {
/** Partition id. */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/PerformanceStatisticsProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/PerformanceStatisticsProcessor.java
index 5d689d754b5..593969d74e5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/PerformanceStatisticsProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/performancestatistics/PerformanceStatisticsProcessor.java
@@ -42,6 +42,7 @@ import org.apache.ignite.internal.util.typedef.internal.A;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteFuture;
import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.internal.processors.metastorage.DistributedMetaStorage.IGNITE_INTERNAL_KEY_PREFIX;
@@ -75,7 +76,7 @@ public class PerformanceStatisticsProcessor extends GridProcessorAdapter {
private final ArrayList<PerformanceStatisticsStateListener> lsnrs = new ArrayList<>();
/** Rotate performance statistics process. */
- private DistributedProcess<Serializable, Serializable> rotateProc;
+ private DistributedProcess<Serializable, Message> rotateProc;
/** Whether performance statistics collection is running. */
private volatile boolean isRunning;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java b/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java
index 8b1ed6955d2..ee0a93945a5 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/DistributedProcess.java
@@ -44,6 +44,7 @@ import org.apache.ignite.internal.util.future.GridFutureAdapter;
import org.apache.ignite.internal.util.typedef.CI3;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
@@ -71,7 +72,7 @@ import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.node2id;
* @see InitMessage
* @see FullMessage
*/
-public class DistributedProcess<I extends Serializable, R extends Serializable> {
+public class DistributedProcess<I extends Serializable, R extends Message> {
/** Process type. */
private final DistributedProcessType type;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/FullMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/FullMessage.java
index eaca62e3594..e302435c9a9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/FullMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/FullMessage.java
@@ -17,13 +17,18 @@
package org.apache.ignite.internal.util.distributed;
-import java.io.Serializable;
+import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.communication.ErrorMessage;
import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage;
import org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType;
+import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.lang.IgniteUuid;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
import org.jetbrains.annotations.Nullable;
/**
@@ -34,24 +39,34 @@ import org.jetbrains.annotations.Nullable;
* @see InitMessage
* @see SingleNodeMessage
*/
-public class FullMessage<R extends Serializable> implements DiscoveryCustomMessage {
+public class FullMessage<R extends Message> implements DiscoveryCustomMessage, Message {
/** Serial version uid. */
private static final long serialVersionUID = 0L;
/** Custom message ID. */
- private final IgniteUuid id = IgniteUuid.randomUuid();
+ @Order(0)
+ IgniteUuid id;
/** Process id. */
- private final UUID processId;
+ @Order(1)
+ UUID processId;
/** Process type. */
- private final int type;
+ @Order(2)
+ int type;
/** Results. */
- private Map<UUID, R> res;
+ @Order(3)
+ Map<UUID, R> res;
/** Errors. */
- private Map<UUID, Throwable> err;
+ @Order(4)
+ Map<UUID, ErrorMessage> err;
+
+ /** Default constructor for {@link MessageFactory}. */
+ public FullMessage() {
+ // No-op.
+ }
/**
* @param processId Process id.
@@ -60,10 +75,18 @@ public class FullMessage<R extends Serializable> implements DiscoveryCustomMessa
* @param err Errors
*/
public FullMessage(UUID processId, DistributedProcessType type, Map<UUID, R> res, Map<UUID, Throwable> err) {
+ this.id = IgniteUuid.randomUuid();
this.processId = processId;
this.type = type.ordinal();
this.res = res;
- this.err = err;
+
+ if (err != null) {
+ HashMap<UUID, ErrorMessage> errMsgs = new HashMap<>();
+
+ err.forEach((id, e) -> errMsgs.put(id, new ErrorMessage(e)));
+
+ this.err = errMsgs;
+ }
}
/** {@inheritDoc} */
@@ -93,7 +116,12 @@ public class FullMessage<R extends Serializable> implements DiscoveryCustomMessa
/** @return Nodes errors. */
public Map<UUID, Throwable> error() {
- return err;
+ return err == null ? null : F.viewReadOnly(err, e -> e.error());
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 30;
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/SingleNodeMessage.java b/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/SingleNodeMessage.java
index 862b2279a9b..b5bfe29d1dd 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/SingleNodeMessage.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/util/distributed/SingleNodeMessage.java
@@ -17,14 +17,13 @@
package org.apache.ignite.internal.util.distributed;
-import java.io.Serializable;
-import java.nio.ByteBuffer;
import java.util.UUID;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.managers.communication.ErrorMessage;
+import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
import org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType;
-import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.plugin.extensions.communication.MessageReader;
-import org.apache.ignite.plugin.extensions.communication.MessageWriter;
+import org.jetbrains.annotations.Nullable;
/**
* Single node result message.
@@ -34,24 +33,29 @@ import org.apache.ignite.plugin.extensions.communication.MessageWriter;
* @see FullMessage
* @see InitMessage
*/
-public class SingleNodeMessage<R extends Serializable> implements Message {
+public class SingleNodeMessage<R extends Message> implements Message {
/** Initial channel message type (value is {@code 176}). */
public static final short TYPE_CODE = 176;
/** Process id. */
- private UUID processId;
+ @Order(0)
+ UUID processId;
/** Process type. */
- private int type;
+ @Order(1)
+ int type;
/** Single node response. */
- private R resp;
+ @Order(value = 2, method = "response")
+ private Message resp;
/** Error. */
- private Throwable err;
+ @Order(3)
+ @Nullable ErrorMessage errMsg;
- /** Empty constructor for marshalling purposes. */
+ /** Default constructor for {@link GridIoMessageFactory}. */
public SingleNodeMessage() {
+ // No-op.
}
/**
@@ -64,92 +68,9 @@ public class SingleNodeMessage<R extends Serializable> implements Message {
this.processId = processId;
this.type = type.ordinal();
this.resp = resp;
- this.err = err;
- }
-
- /** {@inheritDoc} */
- @Override public boolean writeTo(ByteBuffer buf, MessageWriter writer) {
- writer.setBuffer(buf);
-
- if (!writer.isHeaderWritten()) {
- if (!writer.writeHeader(directType()))
- return false;
-
- writer.onHeaderWritten();
- }
-
- switch (writer.state()) {
- case 0:
- if (!writer.writeUuid(processId))
- return false;
-
- writer.incrementState();
-
- case 1:
- if (!writer.writeInt(type))
- return false;
-
- writer.incrementState();
-
- case 2:
- if (!writer.writeByteArray(U.toBytes(resp)))
- return false;
-
- writer.incrementState();
-
- case 3:
- if (!writer.writeByteArray(U.toBytes(err)))
- return false;
-
- writer.incrementState();
- }
-
- return true;
- }
-
- /** {@inheritDoc} */
- @Override public boolean readFrom(ByteBuffer buf, MessageReader reader) {
- reader.setBuffer(buf);
- switch (reader.state()) {
- case 0:
- processId = reader.readUuid();
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- case 1:
- type = reader.readInt();
-
- if (!reader.isLastRead())
- return false;
-
- reader.incrementState();
-
- case 2:
- byte[] dataBytes = reader.readByteArray();
-
- if (!reader.isLastRead())
- return false;
-
- resp = U.fromBytes(dataBytes);
-
- reader.incrementState();
-
- case 3:
- byte[] errBytes = reader.readByteArray();
-
- if (!reader.isLastRead())
- return false;
-
- err = U.fromBytes(errBytes);
-
- reader.incrementState();
- }
-
- return true;
+ if (err != null)
+ errMsg = new ErrorMessage(err);
}
/** {@inheritDoc} */
@@ -169,16 +90,21 @@ public class SingleNodeMessage<R extends Serializable> implements Message {
/** @return Response. */
public R response() {
- return resp;
+ return (R)resp;
+ }
+
+ /** @param resp Response. */
+ public void response(R resp) {
+ this.resp = resp;
}
/** @return {@code True} if finished with error. */
public boolean hasError() {
- return err != null;
+ return errMsg != null;
}
/** @return Error. */
- public Throwable error() {
- return err;
+ @Nullable public Throwable error() {
+ return ErrorMessage.error(errMsg);
}
}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/direct/DirectMarshallingMessagesTest.java b/modules/core/src/test/java/org/apache/ignite/internal/direct/DirectMarshallingMessagesTest.java
deleted file mode 100644
index d3fb2b64dbd..00000000000
--- a/modules/core/src/test/java/org/apache/ignite/internal/direct/DirectMarshallingMessagesTest.java
+++ /dev/null
@@ -1,108 +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.direct;
-
-import java.nio.ByteBuffer;
-import java.util.UUID;
-import java.util.function.Function;
-import org.apache.ignite.internal.managers.communication.GridIoMessageFactory;
-import org.apache.ignite.internal.managers.communication.IgniteMessageFactoryImpl;
-import org.apache.ignite.internal.util.distributed.SingleNodeMessage;
-import org.apache.ignite.internal.util.typedef.internal.U;
-import org.apache.ignite.plugin.extensions.communication.Message;
-import org.apache.ignite.plugin.extensions.communication.MessageFactory;
-import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
-import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.junit.Test;
-
-import static org.apache.ignite.internal.util.distributed.DistributedProcess.DistributedProcessType.TEST_PROCESS;
-import static org.apache.ignite.marshaller.Marshallers.jdk;
-
-/**
- * Messages marshalling test.
- */
-public class DirectMarshallingMessagesTest extends GridCommonAbstractTest {
- /** Message factory. */
- private final MessageFactory msgFactory =
- new IgniteMessageFactoryImpl(new MessageFactoryProvider[] {new GridIoMessageFactory(jdk(), U.gridClassLoader())});
-
- /** */
- @Test
- public void testSingleNodeMessage() {
- SingleNodeMessage<?> srcMsg =
- new SingleNodeMessage<>(UUID.randomUUID(), TEST_PROCESS, "data", new Exception("error"));
-
- SingleNodeMessage<?> resMsg = doMarshalUnmarshal(srcMsg);
-
- assertEquals(srcMsg.type(), resMsg.type());
- assertEquals(srcMsg.processId(), resMsg.processId());
- assertEquals(srcMsg.response(), resMsg.response());
- assertEquals(srcMsg.error().getClass(), resMsg.error().getClass());
- assertEquals(srcMsg.error().getMessage(), resMsg.error().getMessage());
- }
-
- /**
- * @param srcMsg Message to marshal.
- * @param <T> Message type.
- * @return Unmarshalled message.
- */
- private <T extends Message> T doMarshalUnmarshal(T srcMsg) {
- ByteBuffer buf = ByteBuffer.allocate(8 * 1024);
-
- boolean fullyWritten = loopBuffer(buf, 0, buf0 -> srcMsg.writeTo(buf0, new DirectMessageWriter(msgFactory)));
- assertTrue("The message was not written completely.", fullyWritten);
-
- buf.flip();
-
- byte b0 = buf.get();
- byte b1 = buf.get();
-
- short type = (short)((b1 & 0xFF) << 8 | b0 & 0xFF);
-
- assertEquals(srcMsg.directType(), type);
-
- T resMsg = (T)msgFactory.create(type);
-
- boolean fullyRead = loopBuffer(buf, buf.position(),
- buf0 -> resMsg.readFrom(buf0, new DirectMessageReader(msgFactory, null)));
- assertTrue("The message was not read completely.", fullyRead);
-
- return resMsg;
- }
-
- /**
- * @param buf Byte buffer.
- * @param start Start position.
- * @param func Function that is sequentially executed on a different-sized part of the buffer.
- * @return {@code True} if the function returns {@code True} at least once, {@code False} otherwise.
- */
- private boolean loopBuffer(ByteBuffer buf, int start, Function<ByteBuffer, Boolean> func) {
- int pos = start;
-
- do {
- buf.position(start);
- buf.limit(++pos);
-
- if (func.apply(buf))
- return true;
- }
- while (pos < buf.capacity());
-
- return false;
- }
-}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotHandlerTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotHandlerTest.java
index 167c267f790..cd54f29839d 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotHandlerTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IgniteClusterSnapshotHandlerTest.java
@@ -36,6 +36,9 @@ import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree;
+import org.apache.ignite.internal.util.distributed.MessagesPluginProvider;
+import org.apache.ignite.internal.util.distributed.MessagesPluginProvider.MessagesInjectedTcpDiscoverySpi;
+import org.apache.ignite.internal.util.distributed.TestUuidMessage;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.lang.IgniteFuture;
@@ -44,6 +47,7 @@ import org.apache.ignite.plugin.ExtensionRegistry;
import org.apache.ignite.plugin.PluginConfiguration;
import org.apache.ignite.plugin.PluginContext;
import org.apache.ignite.plugin.PluginProvider;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.testframework.GridTestUtils;
import org.jetbrains.annotations.Nullable;
import org.junit.Test;
@@ -71,7 +75,9 @@ public class IgniteClusterSnapshotHandlerTest extends IgniteClusterSnapshotResto
/** {@inheritDoc} */
@Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
- return super.getConfiguration(igniteInstanceName).setPluginProviders(pluginProvider);
+ return super.getConfiguration(igniteInstanceName)
+ .setPluginProviders(pluginProvider, new MessagesPluginProvider())
+ .setDiscoverySpi(new MessagesInjectedTcpDiscoverySpi());
}
/** {@inheritDoc} */
@@ -90,37 +96,37 @@ public class IgniteClusterSnapshotHandlerTest extends IgniteClusterSnapshotResto
AtomicReference<UUID> reqIdRef = new AtomicReference<>();
- handlers.add(new SnapshotHandler<UUID>() {
+ handlers.add(new SnapshotHandler<TestUuidMessage>() {
@Override public SnapshotHandlerType type() {
return SnapshotHandlerType.CREATE;
}
- @Override public UUID invoke(SnapshotHandlerContext ctx) {
- return ctx.metadata().requestId();
+ @Override public TestUuidMessage invoke(SnapshotHandlerContext ctx) {
+ return new TestUuidMessage(ctx.metadata().requestId());
}
@Override public void complete(String name,
- Map<UUID, SnapshotHandlerResult<UUID>> results) throws IgniteCheckedException {
- for (SnapshotHandlerResult<UUID> res : results.values()) {
- if (!reqIdRef.compareAndSet(null, res.data()) && !reqIdRef.get().equals(res.data()))
+ Map<UUID, SnapshotHandlerResult<TestUuidMessage>> results) throws IgniteCheckedException {
+ for (SnapshotHandlerResult<TestUuidMessage> res : results.values()) {
+ if (!reqIdRef.compareAndSet(null, res.data().value()) && !reqIdRef.get().equals(res.data().value()))
throw new IgniteCheckedException("The request ID must be the same on all nodes.");
}
}
});
- handlers.add(new SnapshotHandler<UUID>() {
+ handlers.add(new SnapshotHandler<TestUuidMessage>() {
@Override public SnapshotHandlerType type() {
return SnapshotHandlerType.RESTORE;
}
- @Override public UUID invoke(SnapshotHandlerContext ctx) {
- return ctx.metadata().requestId();
+ @Override public TestUuidMessage invoke(SnapshotHandlerContext ctx) {
+ return new TestUuidMessage(ctx.metadata().requestId());
}
@Override public void complete(String name,
- Map<UUID, SnapshotHandlerResult<UUID>> results) throws IgniteCheckedException {
- for (SnapshotHandlerResult<UUID> res : results.values()) {
- if (!reqIdRef.get().equals(res.data()))
+ Map<UUID, SnapshotHandlerResult<TestUuidMessage>> results) throws IgniteCheckedException {
+ for (SnapshotHandlerResult<TestUuidMessage> res : results.values()) {
+ if (!reqIdRef.get().equals(res.data().value()))
throw new IgniteCheckedException(expMsg);
}
}
@@ -174,12 +180,12 @@ public class IgniteClusterSnapshotHandlerTest extends IgniteClusterSnapshotResto
AtomicBoolean failCreateFlag = new AtomicBoolean(true);
AtomicBoolean failRestoreFlag = new AtomicBoolean(true);
- handlers.add(new SnapshotHandler<Void>() {
+ handlers.add(new SnapshotHandler<>() {
@Override public SnapshotHandlerType type() {
return SnapshotHandlerType.CREATE;
}
- @Override public Void invoke(SnapshotHandlerContext ctx) throws IgniteCheckedException {
+ @Override public Message invoke(SnapshotHandlerContext ctx) throws IgniteCheckedException {
if (failCreateFlag.get())
throw new IgniteCheckedException(expMsg);
@@ -187,12 +193,12 @@ public class IgniteClusterSnapshotHandlerTest extends IgniteClusterSnapshotResto
}
});
- handlers.add(new SnapshotHandler<Void>() {
+ handlers.add(new SnapshotHandler<>() {
@Override public SnapshotHandlerType type() {
return SnapshotHandlerType.RESTORE;
}
- @Override public Void invoke(SnapshotHandlerContext ctx) throws IgniteCheckedException {
+ @Override public Message invoke(SnapshotHandlerContext ctx) throws IgniteCheckedException {
if (failRestoreFlag.get())
throw new IgniteCheckedException(expMsg);
@@ -232,12 +238,12 @@ public class IgniteClusterSnapshotHandlerTest extends IgniteClusterSnapshotResto
*/
@Test
public void testClusterSnapshotHandlerConfigurationMismatch() throws Exception {
- SnapshotHandler<Void> defHnd = new SnapshotHandler<Void>() {
+ SnapshotHandler<Message> defHnd = new SnapshotHandler<>() {
@Override public SnapshotHandlerType type() {
return SnapshotHandlerType.CREATE;
}
- @Override public Void invoke(SnapshotHandlerContext ctx) {
+ @Override public Message invoke(SnapshotHandlerContext ctx) {
return null;
}
};
@@ -263,12 +269,12 @@ public class IgniteClusterSnapshotHandlerTest extends IgniteClusterSnapshotResto
stopGrid(1);
// Case 2: different handlers are loaded on different nodes.
- handlers.add(new SnapshotHandler<Void>() {
+ handlers.add(new SnapshotHandler<>() {
@Override public SnapshotHandlerType type() {
return SnapshotHandlerType.CREATE;
}
- @Nullable @Override public Void invoke(SnapshotHandlerContext ctx) {
+ @Nullable @Override public Message invoke(SnapshotHandlerContext ctx) {
return null;
}
});
@@ -313,16 +319,16 @@ public class IgniteClusterSnapshotHandlerTest extends IgniteClusterSnapshotResto
public void testCrdChangeDuringHandlerCompleteOnSnapshotCreate() throws Exception {
CountDownLatch latch = new CountDownLatch(1);
- handlers.add(new SnapshotHandler<Void>() {
+ handlers.add(new SnapshotHandler<>() {
@Override public SnapshotHandlerType type() {
return SnapshotHandlerType.CREATE;
}
- @Override public Void invoke(SnapshotHandlerContext ctx) {
+ @Override public Message invoke(SnapshotHandlerContext ctx) {
return null;
}
- @Override public void complete(String name, Map<UUID, SnapshotHandlerResult<Void>> results)
+ @Override public void complete(String name, Map<UUID, SnapshotHandlerResult<Message>> results)
throws Exception {
if (latch.getCount() == 1) {
latch.countDown();
@@ -360,12 +366,12 @@ public class IgniteClusterSnapshotHandlerTest extends IgniteClusterSnapshotResto
File snpDir = U.resolveWorkDirectory(U.defaultWorkDirectory(), "ex_snapshots", true);
String expFullPath = new File(snpDir, snpName).getAbsolutePath();
- SnapshotHandler<Void> createHnd = new SnapshotHandler<Void>() {
+ SnapshotHandler<Message> createHnd = new SnapshotHandler<>() {
@Override public SnapshotHandlerType type() {
return SnapshotHandlerType.CREATE;
}
- @Override public Void invoke(SnapshotHandlerContext ctx) {
+ @Override public Message invoke(SnapshotHandlerContext ctx) {
if (!expFullPath.equals(ctx.snapshotFileTree().root().getAbsolutePath()))
throw new IllegalStateException("Expected " + expFullPath + ", actual " + ctx.snapshotFileTree().root());
@@ -373,12 +379,12 @@ public class IgniteClusterSnapshotHandlerTest extends IgniteClusterSnapshotResto
}
};
- SnapshotHandler<Void> restoreHnd = new SnapshotHandler<Void>() {
+ SnapshotHandler<Message> restoreHnd = new SnapshotHandler<>() {
@Override public SnapshotHandlerType type() {
return SnapshotHandlerType.RESTORE;
}
- @Override public Void invoke(SnapshotHandlerContext ctx) throws Exception {
+ @Override public Message invoke(SnapshotHandlerContext ctx) throws Exception {
createHnd.invoke(ctx);
return null;
@@ -411,12 +417,12 @@ public class IgniteClusterSnapshotHandlerTest extends IgniteClusterSnapshotResto
*/
@Test
public void testHandlerExceptionFailSnapshot() throws Exception {
- handlers.add(new SnapshotHandler<Void>() {
+ handlers.add(new SnapshotHandler<>() {
@Override public SnapshotHandlerType type() {
return SnapshotHandlerType.CREATE;
}
- @Override public Void invoke(SnapshotHandlerContext ctx) {
+ @Override public Message invoke(SnapshotHandlerContext ctx) {
// Someone removes snapshot files during creation.
// In this case snapshot must fail.
U.delete(ctx.snapshotFileTree().root());
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotTest.java
index eea0d7b1716..6224135de07 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/snapshot/IncrementalSnapshotTest.java
@@ -31,6 +31,7 @@ import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.managers.communication.ErrorMessage;
import org.apache.ignite.internal.processors.cache.GridLocalConfigManager;
import org.apache.ignite.internal.processors.cache.StoredCacheData;
import org.apache.ignite.internal.processors.cache.persistence.filename.SnapshotFileTree;
@@ -426,7 +427,7 @@ public class IncrementalSnapshotTest extends AbstractSnapshotSelfTest {
DistributedProcess.DistributedProcessType stage = failStage.get();
if (stage != null && stage.ordinal() == singleMsg.type())
- GridTestUtils.setFieldValue(singleMsg, "err", new IgniteException("Test exception."));
+ GridTestUtils.setFieldValue(singleMsg, "errMsg", new ErrorMessage(new IgniteException("Test exception.")));
}
return false;
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/DistributedProcessClientAwaitTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessClientAwaitTest.java
similarity index 89%
rename from modules/core/src/test/java/org/apache/ignite/internal/util/DistributedProcessClientAwaitTest.java
rename to modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessClientAwaitTest.java
index e6e7cc1526c..18bbf5a6d2f 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/util/DistributedProcessClientAwaitTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessClientAwaitTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.util;
+package org.apache.ignite.internal.util.distributed;
import java.util.ArrayList;
import java.util.HashSet;
@@ -28,11 +28,10 @@ import java.util.function.BiFunction;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.GridKernalContext;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
-import org.apache.ignite.internal.util.distributed.DistributedProcess;
-import org.apache.ignite.internal.util.distributed.InitMessage;
-import org.apache.ignite.internal.util.distributed.SingleNodeMessage;
+import org.apache.ignite.internal.util.distributed.MessagesPluginProvider.MessagesInjectedTcpDiscoverySpi;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
@@ -72,6 +71,9 @@ public class DistributedProcessClientAwaitTest extends GridCommonAbstractTest {
cfg.setCommunicationSpi(new TestRecordingCommunicationSpi());
+ cfg.setPluginProviders(new MessagesPluginProvider());
+ cfg.setDiscoverySpi(new MessagesInjectedTcpDiscoverySpi());
+
return cfg;
}
@@ -102,7 +104,7 @@ public class DistributedProcessClientAwaitTest extends GridCommonAbstractTest {
public void testSkipWaitingFailedClient() throws Exception {
finishLatchRef.set(new CountDownLatch(NODES_CNT));
- List<DistributedProcess<Integer, Integer>> processes = new ArrayList<>(NODES_CNT + 1);
+ List<DistributedProcess<Integer, Message>> processes = new ArrayList<>(NODES_CNT + 1);
TestRecordingCommunicationSpi clnCommSpi = TestRecordingCommunicationSpi.spi(grid(NODES_CNT));
clnCommSpi.blockMessages((node, msg) -> msg instanceof SingleNodeMessage);
@@ -113,7 +115,7 @@ public class DistributedProcessClientAwaitTest extends GridCommonAbstractTest {
nodeIdsRes.add(grid(i).localNode().id());
for (int n = 0; n < NODES_CNT + 1; n++) {
- DistributedProcess<Integer, Integer> dp = new TestDistributedProcess(
+ DistributedProcess<Integer, Message> dp = new TestDistributedProcess(
nodeIdsRes, grid(n).context(), (id, req) -> new InitMessage<>(id, TEST_PROCESS, req, true));
processes.add(dp);
@@ -135,7 +137,7 @@ public class DistributedProcessClientAwaitTest extends GridCommonAbstractTest {
public void testChangedCoordinatorAwaitsClientResult() throws Exception {
finishLatchRef.set(new CountDownLatch(NODES_CNT));
- List<DistributedProcess<Integer, Integer>> processes = new ArrayList<>(NODES_CNT + 1);
+ List<DistributedProcess<Integer, Message>> processes = new ArrayList<>(NODES_CNT + 1);
TestRecordingCommunicationSpi clnCommSpi = TestRecordingCommunicationSpi.spi(grid(NODES_CNT));
clnCommSpi.blockMessages((node, msg) -> msg instanceof SingleNodeMessage);
@@ -146,7 +148,7 @@ public class DistributedProcessClientAwaitTest extends GridCommonAbstractTest {
nodeIdsRes.add(grid(i).localNode().id());
for (int n = 0; n < NODES_CNT + 1; n++) {
- DistributedProcess<Integer, Integer> dp = new TestDistributedProcess(
+ DistributedProcess<Integer, Message> dp = new TestDistributedProcess(
nodeIdsRes, grid(n).context(), (id, req) -> new InitMessage<>(id, TEST_PROCESS, req, true));
processes.add(dp);
@@ -172,10 +174,10 @@ public class DistributedProcessClientAwaitTest extends GridCommonAbstractTest {
Set<UUID> expNodeIdRes,
BiFunction<UUID, Integer, ? extends InitMessage<Integer>> initMsgFactory
) throws Exception {
- List<DistributedProcess<Integer, Integer>> processes = new ArrayList<>(NODES_CNT + 1);
+ List<DistributedProcess<Integer, Message>> processes = new ArrayList<>(NODES_CNT + 1);
for (int n = 0; n < NODES_CNT + 1; n++) {
- DistributedProcess<Integer, Integer> dp = new TestDistributedProcess(
+ DistributedProcess<Integer, Message> dp = new TestDistributedProcess(
expNodeIdRes, grid(n).context(), initMsgFactory);
processes.add(dp);
@@ -194,7 +196,7 @@ public class DistributedProcessClientAwaitTest extends GridCommonAbstractTest {
}
/** */
- private static class TestDistributedProcess extends DistributedProcess<Integer, Integer> {
+ private static class TestDistributedProcess extends DistributedProcess<Integer, Message> {
/** */
public TestDistributedProcess(
Set<UUID> expNodeIdsRes,
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/DistributedProcessCoordinatorLeftTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessCoordinatorLeftTest.java
similarity index 88%
rename from modules/core/src/test/java/org/apache/ignite/internal/util/DistributedProcessCoordinatorLeftTest.java
rename to modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessCoordinatorLeftTest.java
index d4f696106a8..13203ae8bee 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/util/DistributedProcessCoordinatorLeftTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessCoordinatorLeftTest.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.util;
+package org.apache.ignite.internal.util.distributed;
import java.util.Collections;
import java.util.HashMap;
@@ -28,7 +28,7 @@ import org.apache.ignite.failure.FailureContext;
import org.apache.ignite.failure.FailureHandler;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.util.distributed.DistributedProcess;
+import org.apache.ignite.internal.util.distributed.MessagesPluginProvider.MessagesInjectedTcpDiscoverySpi;
import org.apache.ignite.internal.util.typedef.G;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
@@ -86,6 +86,9 @@ public class DistributedProcessCoordinatorLeftTest extends GridCommonAbstractTes
}
});
+ cfg.setPluginProviders(new MessagesPluginProvider());
+ cfg.setDiscoverySpi(new MessagesInjectedTcpDiscoverySpi());
+
return cfg;
}
@@ -109,14 +112,15 @@ public class DistributedProcessCoordinatorLeftTest extends GridCommonAbstractTes
CountDownLatch startLatch = new CountDownLatch(NODES_CNT);
CountDownLatch finishLatch = new CountDownLatch(NODES_CNT - 1);
- HashMap<String, DistributedProcess<Integer, Integer>> processes = new HashMap<>();
+ HashMap<String, DistributedProcess<Integer, TestIntegerMessage>> processes = new HashMap<>();
int procRes = 1;
for (Ignite grid : G.allGrids()) {
- DistributedProcess<Integer, Integer> dp = new DistributedProcess<>(((IgniteEx)grid).context(), TEST_PROCESS,
+ DistributedProcess<Integer, TestIntegerMessage> dp = new DistributedProcess<>(((IgniteEx)grid).context(),
+ TEST_PROCESS,
req -> {
- IgniteInternalFuture<Integer> fut = runAsync(() -> {
+ IgniteInternalFuture<TestIntegerMessage> fut = runAsync(() -> {
try {
nodeLeftLatch.await();
}
@@ -124,7 +128,7 @@ public class DistributedProcessCoordinatorLeftTest extends GridCommonAbstractTes
fail("Unexpected interrupt.");
}
- return req;
+ return new TestIntegerMessage(req);
});
// A single message will be sent before this latch released.
@@ -137,7 +141,7 @@ public class DistributedProcessCoordinatorLeftTest extends GridCommonAbstractTes
return fut;
},
(uuid, res, err) -> {
- if (res.values().size() == NODES_CNT - 1 && res.values().stream().allMatch(i -> i == procRes))
+ if (res.values().size() == NODES_CNT - 1 && res.values().stream().allMatch(i -> i.value() == procRes))
finishLatch.countDown();
else
fail("Unexpected process result [res=" + res + ", err=" + err + ']');
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/DistributedProcessErrorHandlingTest.java b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessErrorHandlingTest.java
similarity index 94%
rename from modules/core/src/test/java/org/apache/ignite/internal/util/DistributedProcessErrorHandlingTest.java
rename to modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessErrorHandlingTest.java
index 08292c01d6d..7254c47cdb4 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/util/DistributedProcessErrorHandlingTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/DistributedProcessErrorHandlingTest.java
@@ -15,22 +15,21 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.util;
+package org.apache.ignite.internal.util.distributed;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.function.BiFunction;
-
import org.apache.ignite.Ignite;
import org.apache.ignite.cluster.ClusterState;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.failure.StopNodeFailureHandler;
import org.apache.ignite.internal.IgniteEx;
-import org.apache.ignite.internal.util.distributed.DistributedProcess;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.plugin.extensions.communication.Message;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.junit.Test;
import org.junit.runner.RunWith;
@@ -84,7 +83,7 @@ public class DistributedProcessErrorHandlingTest extends GridCommonAbstractTest
req -> runAsync(() -> {
failOnNode(ign); // Fails processing request in a spawned thread.
- return 0;
+ return null;
}),
(id, res, err) -> {
if (failClient)
@@ -107,7 +106,7 @@ public class DistributedProcessErrorHandlingTest extends GridCommonAbstractTest
req -> {
failOnNode(ign); // Fails processing request in the discovery thread.
- return new GridFinishedFuture<>(0);
+ return new GridFinishedFuture<>();
},
(id, res, err) -> {
if (failClient)
@@ -127,7 +126,7 @@ public class DistributedProcessErrorHandlingTest extends GridCommonAbstractTest
public void testFinishFailureHandled() throws Exception {
checkDistributedProcess((ign, latch) ->
new DistributedProcess<>(ign.context(), TEST_PROCESS,
- req -> new GridFinishedFuture<>(0),
+ req -> new GridFinishedFuture<>(),
(uuid, res, err) -> {
assertEquals(SRV_NODES, res.values().size());
latch.countDown();
@@ -138,9 +137,9 @@ public class DistributedProcessErrorHandlingTest extends GridCommonAbstractTest
/** */
private void checkDistributedProcess(
- BiFunction<IgniteEx, CountDownLatch, DistributedProcess<Integer, Integer>> processFactory
+ BiFunction<IgniteEx, CountDownLatch, DistributedProcess<Integer, Message>> processFactory
) throws Exception {
- DistributedProcess<Integer, Integer> proc = null;
+ DistributedProcess<Integer, Message> proc = null;
CountDownLatch latch = new CountDownLatch(SRV_NODES + 1);
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/MessagesPluginProvider.java b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/MessagesPluginProvider.java
new file mode 100644
index 00000000000..203d91ad0f9
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/MessagesPluginProvider.java
@@ -0,0 +1,64 @@
+/*
+ * 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.util.distributed;
+
+import org.apache.ignite.internal.managers.communication.IgniteMessageFactoryImpl;
+import org.apache.ignite.internal.managers.discovery.DiscoveryMessageFactory;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.plugin.AbstractTestPluginProvider;
+import org.apache.ignite.plugin.ExtensionRegistry;
+import org.apache.ignite.plugin.PluginContext;
+import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.testframework.GridTestUtils;
+
+import static org.apache.ignite.marshaller.Marshallers.jdk;
+import static org.junit.Assert.assertTrue;
+
+/** */
+public class MessagesPluginProvider extends AbstractTestPluginProvider {
+ /** */
+ private static final MessageFactoryProvider FACTORY_PROVIDER = f -> {
+ f.register((short)10_000, TestIntegerMessage::new, new TestIntegerMessageSerializer());
+ f.register((short)10_001, TestUuidMessage::new, new TestUuidMessageSerializer());
+ };
+
+ /** {@inheritDoc} */
+ @Override public String name() {
+ return "Distributed process test messages plugin";
+ }
+
+ /** {@inheritDoc} */
+ @Override public void initExtensions(PluginContext ctx, ExtensionRegistry registry) {
+ registry.registerExtension(MessageFactoryProvider.class, FACTORY_PROVIDER);
+
+ // Register messages into the DiscoverySpi.
+ assertTrue(ctx.igniteConfiguration().getDiscoverySpi() instanceof MessagesInjectedTcpDiscoverySpi);
+ }
+
+ /** */
+ public static class MessagesInjectedTcpDiscoverySpi extends TcpDiscoverySpi {
+ /** {@inheritDoc} */
+ @Override protected void initLocalNode(int srvPort, boolean addExtAddrAttr) {
+ GridTestUtils.setFieldValue(this, TcpDiscoverySpi.class, "msgFactory", new IgniteMessageFactoryImpl(
+ new MessageFactoryProvider[] { new DiscoveryMessageFactory(jdk(), U.gridClassLoader()), FACTORY_PROVIDER}));
+
+ super.initLocalNode(srvPort, addExtAddrAttr);
+ }
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/TestIntegerMessage.java b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/TestIntegerMessage.java
new file mode 100644
index 00000000000..84ff66e8a8e
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/TestIntegerMessage.java
@@ -0,0 +1,49 @@
+/*
+ * 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.util.distributed;
+
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+
+/** */
+public class TestIntegerMessage implements Message {
+ /** */
+ @Order(0)
+ int val;
+
+ /** Default constructor for {@link MessageFactory}. */
+ public TestIntegerMessage() {
+ // No-op.
+ }
+
+ /** */
+ public TestIntegerMessage(int val) {
+ this.val = val;
+ }
+
+ /** */
+ public int value() {
+ return val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 10_000;
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/TestUuidMessage.java b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/TestUuidMessage.java
new file mode 100644
index 00000000000..0cf954b1a9e
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/util/distributed/TestUuidMessage.java
@@ -0,0 +1,50 @@
+/*
+ * 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.util.distributed;
+
+import java.util.UUID;
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.plugin.extensions.communication.MessageFactory;
+
+/** */
+public class TestUuidMessage implements Message {
+ /** */
+ @Order(0)
+ UUID val;
+
+ /** Default constructor for {@link MessageFactory}. */
+ public TestUuidMessage() {
+ // No-op.
+ }
+
+ /** */
+ public TestUuidMessage(UUID val) {
+ this.val = val;
+ }
+
+ /** */
+ public UUID value() {
+ return val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public short directType() {
+ return 10_001;
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteMarshallerSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteMarshallerSelfTestSuite.java
index 3872d67eec6..aa268afe0fa 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteMarshallerSelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteMarshallerSelfTestSuite.java
@@ -17,7 +17,6 @@
package org.apache.ignite.testsuites;
-import org.apache.ignite.internal.direct.DirectMarshallingMessagesTest;
import org.apache.ignite.internal.direct.stream.DirectByteBufferStreamImplByteOrderSelfTest;
import org.apache.ignite.internal.marshaller.optimized.OptimizedMarshallerEnumSelfTest;
import org.apache.ignite.internal.marshaller.optimized.OptimizedMarshallerPooledSelfTest;
@@ -51,7 +50,6 @@ import org.junit.runners.Suite;
GridHandleTableSelfTest.class,
OptimizedMarshallerPooledSelfTest.class,
MarshallerEnumDeadlockMultiJvmTest.class,
- DirectMarshallingMessagesTest.class,
ObjectInputStreamFilteringTest.class,
})
public class IgniteMarshallerSelfTestSuite {
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java
index 30a20c00d97..c06cbfe8fd4 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteUtilSelfTestSuite.java
@@ -25,9 +25,6 @@ import org.apache.ignite.internal.thread.pool.GridThreadPoolExecutorServiceSelfT
import org.apache.ignite.internal.thread.pool.IgniteThreadPoolSizeTest;
import org.apache.ignite.internal.thread.pool.ThreadPoolMetricsTest;
import org.apache.ignite.internal.util.BasicRateLimiterTest;
-import org.apache.ignite.internal.util.DistributedProcessClientAwaitTest;
-import org.apache.ignite.internal.util.DistributedProcessCoordinatorLeftTest;
-import org.apache.ignite.internal.util.DistributedProcessErrorHandlingTest;
import org.apache.ignite.internal.util.GridArraysSelfTest;
import org.apache.ignite.internal.util.GridConcurrentMultiPairQueueTest;
import org.apache.ignite.internal.util.GridCountDownCallbackTest;
@@ -36,6 +33,9 @@ import org.apache.ignite.internal.util.IgniteDevOnlyLogTest;
import org.apache.ignite.internal.util.IgniteExceptionRegistrySelfTest;
import org.apache.ignite.internal.util.IgniteUtilsSelfTest;
import org.apache.ignite.internal.util.IgniteUtilsUnitTest;
+import org.apache.ignite.internal.util.distributed.DistributedProcessClientAwaitTest;
+import org.apache.ignite.internal.util.distributed.DistributedProcessCoordinatorLeftTest;
+import org.apache.ignite.internal.util.distributed.DistributedProcessErrorHandlingTest;
import org.apache.ignite.internal.util.nio.GridNioDelimitedBufferSelfTest;
import org.apache.ignite.internal.util.nio.GridNioSelfTest;
import org.apache.ignite.internal.util.nio.GridNioServerTest;