Skip to content

Commit d9046e5

Browse files
NSAmelchevNSAmelchev
committed
IGNITE-27631 Use MessageSerializer for SingleNodeMessage
1 parent 6cca09c commit d9046e5

41 files changed

Lines changed: 1163 additions & 488 deletions

File tree

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎modules/core/pom.xml‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -35,6 +35,7 @@
3535
<properties>
3636
<ignite.update.notifier.product>apache-ignite</ignite.update.notifier.product>
3737
<ignite.generated.source.path>${project.build.directory}/generated-sources/codegen</ignite.generated.source.path>
38+
<ignite.generated.test.source.path>${project.build.directory}/generated-test-sources/codegen</ignite.generated.test.source.path>
3839
</properties>
3940

4041
<dependencies>
@@ -376,13 +377,26 @@
376377
</sources>
377378
</configuration>
378379
</execution>
380+
<execution>
381+
<id>add-generated-test-sources</id>
382+
<phase>generate-test-sources</phase>
383+
<goals>
384+
<goal>add-test-source</goal>
385+
</goals>
386+
<configuration>
387+
<sources>
388+
<source>${ignite.generated.test.source.path}</source>
389+
</sources>
390+
</configuration>
391+
</execution>
379392
</executions>
380393
</plugin>
381394

382395
<plugin>
383396
<artifactId>maven-compiler-plugin</artifactId>
384397
<configuration>
385398
<generatedSourcesDirectory>${ignite.generated.source.path}</generatedSourcesDirectory>
399+
<generatedTestSourcesDirectory>${ignite.generated.test.source.path}</generatedTestSourcesDirectory>
386400
<annotationProcessors>
387401
<annotationProcessor>org.apache.ignite.internal.MessageProcessor</annotationProcessor>
388402
<annotationProcessor>org.apache.ignite.internal.idto.IgniteDataTransferObjectProcessor</annotationProcessor>

‎modules/core/src/main/java/org/apache/ignite/internal/managers/communication/GridIoMessageFactory.java‎

Lines changed: 40 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -231,12 +231,34 @@
231231
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequestSerializer;
232232
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponse;
233233
import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponseSerializer;
234+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.DataStreamerUpdatesHandlerResult;
235+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.DataStreamerUpdatesHandlerResultSerializer;
234236
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotAwareMessage;
235237
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotAwareMessageSerializer;
238+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotVerifyResult;
239+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotVerifyResultMarshallableSerializer;
240+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersNodeResponse;
241+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersNodeResponseSerializer;
242+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersResponse;
243+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersResponseSerializer;
244+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckPartitionHashesResponse;
245+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckPartitionHashesResponseMarshallableSerializer;
246+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckResponse;
247+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckResponseSerializer;
236248
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFilesFailureMessage;
237249
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFilesFailureMessageSerializer;
238250
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFilesRequestMessage;
239251
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotFilesRequestMessageSerializer;
252+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandlerResult;
253+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandlerResultSerializer;
254+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataResponse;
255+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataResponseMarshallableSerializer;
256+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperationResponse;
257+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperationResponseSerializer;
258+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyHandlerResponse;
259+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyHandlerResponseMarshallableSerializer;
260+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreOperationResponse;
261+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreOperationResponseMarshallableSerializer;
240262
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryRequest;
241263
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryRequestSerializer;
242264
import org.apache.ignite.internal.processors.cache.query.GridCacheQueryResponse;
@@ -332,6 +354,7 @@
332354
import org.apache.ignite.internal.util.UUIDCollectionMessage;
333355
import org.apache.ignite.internal.util.UUIDCollectionMessageSerializer;
334356
import org.apache.ignite.internal.util.distributed.SingleNodeMessage;
357+
import org.apache.ignite.internal.util.distributed.SingleNodeMessageSerializer;
335358
import org.apache.ignite.marshaller.Marshaller;
336359
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
337360
import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
@@ -519,7 +542,7 @@ public GridIoMessageFactory(Marshaller cstDataMarshall, ClassLoader cstDataMarsh
519542
factory.register(GridQueryKillResponse.TYPE_CODE, GridQueryKillResponse::new, new GridQueryKillResponseSerializer());
520543
factory.register(GridIoSecurityAwareMessage.TYPE_CODE, GridIoSecurityAwareMessage::new, new GridIoSecurityAwareMessageSerializer());
521544
factory.register(SessionChannelMessage.TYPE_CODE, SessionChannelMessage::new, new SessionChannelMessageSerializer());
522-
factory.register(SingleNodeMessage.TYPE_CODE, SingleNodeMessage::new);
545+
factory.register(SingleNodeMessage.TYPE_CODE, SingleNodeMessage::new, new SingleNodeMessageSerializer());
523546
factory.register((short)177, TcpInverseConnectionResponseMessage::new, new TcpInverseConnectionResponseMessageSerializer());
524547
factory.register(SnapshotFilesRequestMessage.TYPE_CODE, SnapshotFilesRequestMessage::new,
525548
new SnapshotFilesRequestMessageSerializer());
@@ -563,6 +586,22 @@ public GridIoMessageFactory(Marshaller cstDataMarshall, ClassLoader cstDataMarsh
563586
factory.register(GridPartitionStateMap.TYPE_CODE, GridPartitionStateMap::new, new GridPartitionStateMapSerializer());
564587
factory.register(GridDhtPartitionMap.TYPE_CODE, GridDhtPartitionMap::new, new GridDhtPartitionMapSerializer());
565588
factory.register(GridDhtPartitionFullMap.TYPE_CODE, GridDhtPartitionFullMap::new, new GridDhtPartitionFullMapSerializer());
589+
factory.register((short)520, SnapshotOperationResponse::new, new SnapshotOperationResponseSerializer());
590+
factory.register((short)521, SnapshotHandlerResult::new, new SnapshotHandlerResultSerializer());
591+
factory.register((short)522, DataStreamerUpdatesHandlerResult::new, new DataStreamerUpdatesHandlerResultSerializer());
592+
factory.register((short)523, SnapshotCheckResponse::new, new SnapshotCheckResponseSerializer());
593+
factory.register((short)524, IncrementalSnapshotVerifyResult::new,
594+
new IncrementalSnapshotVerifyResultMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
595+
factory.register((short)525, SnapshotRestoreOperationResponse::new,
596+
new SnapshotRestoreOperationResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
597+
factory.register((short)526, SnapshotMetadataResponse::new,
598+
new SnapshotMetadataResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
599+
factory.register((short)527, SnapshotCheckPartitionHashesResponse::new,
600+
new SnapshotCheckPartitionHashesResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
601+
factory.register((short)528, SnapshotCheckHandlersResponse::new, new SnapshotCheckHandlersResponseSerializer());
602+
factory.register((short)529, SnapshotCheckHandlersNodeResponse::new, new SnapshotCheckHandlersNodeResponseSerializer());
603+
factory.register((short)530, SnapshotPartitionsVerifyHandlerResponse::new,
604+
new SnapshotPartitionsVerifyHandlerResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
566605

567606
// [-3..119] [124..129] [-23..-28] [-36..-55] [183..188] - this
568607
// [120..123] - DR

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

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -45,6 +45,30 @@
4545
import org.apache.ignite.internal.processors.cache.binary.MetadataRemoveProposedMessageSerializer;
4646
import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateAcceptedMessage;
4747
import org.apache.ignite.internal.processors.cache.binary.MetadataUpdateAcceptedMessageSerializer;
48+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.DataStreamerUpdatesHandlerResult;
49+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.DataStreamerUpdatesHandlerResultSerializer;
50+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotVerifyResult;
51+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.IncrementalSnapshotVerifyResultMarshallableSerializer;
52+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersNodeResponse;
53+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersNodeResponseSerializer;
54+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersResponse;
55+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckHandlersResponseSerializer;
56+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckPartitionHashesResponse;
57+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckPartitionHashesResponseMarshallableSerializer;
58+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckResponse;
59+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotCheckResponseSerializer;
60+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandlerResult;
61+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotHandlerResultSerializer;
62+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataResponse;
63+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotMetadataResponseMarshallableSerializer;
64+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperationResponse;
65+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotOperationResponseSerializer;
66+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyHandlerResponse;
67+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotPartitionsVerifyHandlerResponseMarshallableSerializer;
68+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreOperationResponse;
69+
import org.apache.ignite.internal.processors.cache.persistence.snapshot.SnapshotRestoreOperationResponseMarshallableSerializer;
70+
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
71+
import org.apache.ignite.internal.processors.cache.version.GridCacheVersionSerializer;
4872
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessage;
4973
import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateFinishMessageSerializer;
5074
import org.apache.ignite.internal.processors.continuous.StopRoutineAckDiscoveryMessage;
@@ -81,6 +105,8 @@
81105
import org.apache.ignite.internal.processors.query.schema.operation.SchemaIndexCreateOperationMarshallableSerializer;
82106
import org.apache.ignite.internal.processors.query.schema.operation.SchemaIndexDropOperation;
83107
import org.apache.ignite.internal.processors.query.schema.operation.SchemaIndexDropOperationSerializer;
108+
import org.apache.ignite.internal.util.distributed.FullMessage;
109+
import org.apache.ignite.internal.util.distributed.FullMessageSerializer;
84110
import org.apache.ignite.marshaller.Marshaller;
85111
import org.apache.ignite.plugin.extensions.communication.MessageFactory;
86112
import org.apache.ignite.plugin.extensions.communication.MessageFactoryProvider;
@@ -232,6 +258,9 @@ public DiscoveryMessageFactory(Marshaller cstDataMarshall, ClassLoader cstDataMa
232258
new TcpDiscoveryClientReconnectMessageMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
233259
factory.register((short)29, TcpDiscoveryNodeAddedMessage::new,
234260
new TcpDiscoveryNodeAddedMessageMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
261+
factory.register((short)30, FullMessage::new, new FullMessageSerializer());
262+
263+
factory.register((short)86, GridCacheVersion::new, new GridCacheVersionSerializer());
235264

236265
// DiscoveryCustomMessage
237266
factory.register((short)500, CacheStatisticsModeChangeMessage::new, new CacheStatisticsModeChangeMessageSerializer());
@@ -257,5 +286,21 @@ public DiscoveryMessageFactory(Marshaller cstDataMarshall, ClassLoader cstDataMa
257286
factory.register((short)518, MappingProposedMessage::new, new MappingProposedMessageSerializer());
258287
factory.register((short)519, MarshallerMappingItem::new, new MarshallerMappingItemSerializer());
259288

289+
factory.register((short)520, SnapshotOperationResponse::new, new SnapshotOperationResponseSerializer());
290+
factory.register((short)521, SnapshotHandlerResult::new, new SnapshotHandlerResultSerializer());
291+
factory.register((short)522, DataStreamerUpdatesHandlerResult::new, new DataStreamerUpdatesHandlerResultSerializer());
292+
factory.register((short)523, SnapshotCheckResponse::new, new SnapshotCheckResponseSerializer());
293+
factory.register((short)524, IncrementalSnapshotVerifyResult::new,
294+
new IncrementalSnapshotVerifyResultMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
295+
factory.register((short)525, SnapshotRestoreOperationResponse::new,
296+
new SnapshotRestoreOperationResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
297+
factory.register((short)526, SnapshotMetadataResponse::new,
298+
new SnapshotMetadataResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
299+
factory.register((short)527, SnapshotCheckPartitionHashesResponse::new,
300+
new SnapshotCheckPartitionHashesResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
301+
factory.register((short)528, SnapshotCheckHandlersResponse::new, new SnapshotCheckHandlersResponseSerializer());
302+
factory.register((short)529, SnapshotCheckHandlersNodeResponse::new, new SnapshotCheckHandlersNodeResponseSerializer());
303+
factory.register((short)530, SnapshotPartitionsVerifyHandlerResponse::new,
304+
new SnapshotPartitionsVerifyHandlerResponseMarshallableSerializer(cstDataMarshall, cstDataMarshallClsLdr));
260305
}
261306
}

‎modules/core/src/main/java/org/apache/ignite/internal/managers/encryption/GridEncryptionManager.java‎

Lines changed: 9 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,7 @@
7171
import org.apache.ignite.lang.IgniteFuture;
7272
import org.apache.ignite.lang.IgniteFutureCancelledException;
7373
import org.apache.ignite.lang.IgniteUuid;
74+
import org.apache.ignite.plugin.extensions.communication.Message;
7475
import org.apache.ignite.spi.IgniteNodeValidationResult;
7576
import org.apache.ignite.spi.IgniteSpiException;
7677
import org.apache.ignite.spi.discovery.DiscoveryDataBag;
@@ -203,10 +204,10 @@ public class GridEncryptionManager extends GridManagerAdapter<EncryptionSpi> imp
203204
* Master key change prepare process. Checks that all server nodes have the same new master key and then starts
204205
* finish process.
205206
*/
206-
private DistributedProcess<MasterKeyChangeRequest, EmptyResult> prepareMKChangeProc;
207+
private DistributedProcess<MasterKeyChangeRequest, Message> prepareMKChangeProc;
207208

208209
/** Process to perform the master key change. Changes master key and reencrypt group keys. */
209-
private DistributedProcess<MasterKeyChangeRequest, EmptyResult> performMKChangeProc;
210+
private DistributedProcess<MasterKeyChangeRequest, Message> performMKChangeProc;
210211

211212
/**
212213
* A two-phase distributed process that rotates the encryption keys of specified cache groups and initiates
@@ -1488,7 +1489,7 @@ public void applyReencryptionStartRecord(ReencryptionStartRecord rec) {
14881489
* @param req Request.
14891490
* @return Result future.
14901491
*/
1491-
private IgniteInternalFuture<EmptyResult> prepareMasterKeyChange(MasterKeyChangeRequest req) {
1492+
private IgniteInternalFuture<Message> prepareMasterKeyChange(MasterKeyChangeRequest req) {
14921493
if (masterKeyChangeRequest != null) {
14931494
return new GridFinishedFuture<>(new IgniteException("Master key change was rejected. " +
14941495
"The previous change was not completed."));
@@ -1524,7 +1525,7 @@ private IgniteInternalFuture<EmptyResult> prepareMasterKeyChange(MasterKeyChange
15241525
ctx.localNodeId() + ']', e));
15251526
}
15261527

1527-
return new GridFinishedFuture<>(new EmptyResult());
1528+
return new GridFinishedFuture<>();
15281529
}
15291530

15301531
/**
@@ -1534,7 +1535,7 @@ private IgniteInternalFuture<EmptyResult> prepareMasterKeyChange(MasterKeyChange
15341535
* @param res Results.
15351536
* @param err Errors.
15361537
*/
1537-
private void finishPrepareMasterKeyChange(UUID id, Map<UUID, EmptyResult> res, Map<UUID, Throwable> err) {
1538+
private void finishPrepareMasterKeyChange(UUID id, Map<UUID, Message> res, Map<UUID, Throwable> err) {
15381539
if (!err.isEmpty()) {
15391540
if (masterKeyChangeRequest != null && masterKeyChangeRequest.requestId().equals(id))
15401541
masterKeyChangeRequest = null;
@@ -1551,7 +1552,7 @@ else if (U.isLocalNodeCoordinator(ctx.discovery()))
15511552
* @param req Request.
15521553
* @return Result future.
15531554
*/
1554-
private IgniteInternalFuture<EmptyResult> performMasterKeyChange(MasterKeyChangeRequest req) {
1555+
private IgniteInternalFuture<Message> performMasterKeyChange(MasterKeyChangeRequest req) {
15551556
if (masterKeyChangeRequest == null || !masterKeyChangeRequest.equals(req))
15561557
return new GridFinishedFuture<>(new IgniteException("Unknown master key change was rejected."));
15571558

@@ -1569,7 +1570,7 @@ private IgniteInternalFuture<EmptyResult> performMasterKeyChange(MasterKeyChange
15691570

15701571
masterKeyDigest = req.digest();
15711572

1572-
return new GridFinishedFuture<>(new EmptyResult());
1573+
return new GridFinishedFuture<>();
15731574
}
15741575

15751576
/**
@@ -1579,7 +1580,7 @@ private IgniteInternalFuture<EmptyResult> performMasterKeyChange(MasterKeyChange
15791580
* @param res Results.
15801581
* @param err Errors.
15811582
*/
1582-
private void finishPerformMasterKeyChange(UUID id, Map<UUID, EmptyResult> res, Map<UUID, Throwable> err) {
1583+
private void finishPerformMasterKeyChange(UUID id, Map<UUID, Message> res, Map<UUID, Throwable> err) {
15831584
completeMasterKeyChangeFuture(id, err);
15841585
}
15851586

@@ -1818,12 +1819,6 @@ byte[] digest() {
18181819
}
18191820
}
18201821

1821-
/** */
1822-
protected static class EmptyResult implements Serializable {
1823-
/** Serial version uid. */
1824-
private static final long serialVersionUID = 0L;
1825-
}
1826-
18271822
/** */
18281823
protected static class NodeEncryptionKeys implements Serializable {
18291824
/** */

0 commit comments

Comments
 (0)