You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "ibessonov (via GitHub)" <gi...@apache.org> on 2023/10/31 13:55:58 UTC
[PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
ibessonov opened a new pull request, #2772:
URL: https://github.com/apache/ignite-3/pull/2772
https://issues.apache.org/jira/browse/IGNITE-19453
Thank you for submitting the pull request.
To streamline the review process of the patch and ensure better code quality
we ask both an author and a reviewer to verify the following:
### The Review Checklist
- [ ] **Formal criteria:** TC status, codestyle, mandatory documentation. Also make sure to complete the following:
\- There is a single JIRA ticket related to the pull request.
\- The web-link to the pull request is attached to the JIRA ticket.
\- The JIRA ticket has the Patch Available state.
\- The description of the JIRA ticket explains WHAT was made, WHY and HOW.
\- The pull request title is treated as the final commit message. The following pattern must be used: IGNITE-XXXX Change summary where XXXX - number of JIRA issue.
- [ ] **Design:** new code conforms with the design principles of the components it is added to.
- [ ] **Patch quality:** patch cannot be split into smaller pieces, its size must be reasonable.
- [ ] **Code quality:** code is clean and readable, necessary developer documentation is added if needed.
- [ ] **Tests code quality:** test set covers positive/negative scenarios, happy/edge cases. Tests are effective in terms of execution time and resources.
### Notes
- [Apache Ignite Coding Guidelines](https://cwiki.apache.org/confluence/display/IGNITE/Java+Code+Style+Guide)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1377665687
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/ActionRequest.java:
##########
@@ -32,9 +34,15 @@ public interface ActionRequest extends Message {
String groupId();
/**
- * @return Action's command.
+ * @return Serialized action's command. Specific serialization format may differ from group to group.
+ */
+ byte[] command();
Review Comment:
Question to myself. Maybe we should have a setter for this property as well. See `patchCommandBeforeApply`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392288647
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/service/BeforeApplyHandler.java:
##########
@@ -38,6 +38,7 @@ public interface BeforeApplyHandler {
* this is a place to do it.
*
* @param command The command.
+ * @return {@code true} if command has been changed, {@code false} otherwise.
Review Comment:
I did
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392200443
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftManager.java:
##########
@@ -148,6 +150,7 @@ <T extends RaftGroupService> CompletableFuture<T> startRaftGroupNodeAndWaitNodeR
* @return Future that will be completed with an instance of a Raft group service.
* @throws NodeStoppingException If node stopping intention was detected.
*/
+ @TestOnly
Review Comment:
Wait a minute, it's only used in tests, the annotation is correct...
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392183206
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/service/BeforeApplyHandler.java:
##########
@@ -38,6 +38,7 @@ public interface BeforeApplyHandler {
* this is a place to do it.
*
* @param command The command.
+ * @return {@code true} if command has been changed, {@code false} otherwise.
Review Comment:
It means that some fields in command instance have been modified.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392217891
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/RaftGroupServiceImpl.java:
##########
@@ -123,6 +125,7 @@ public class RaftGroupServiceImpl implements RaftGroupService {
* @param membersConfiguration Raft members configuration.
* @param leader Group leader.
* @param executor Executor for retrying requests.
+ * @param commandsMarshaller Marshaller that should be used to [de]serialize commands.
Review Comment:
It would be great, but at your discretion.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392210336
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/Marshaller.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.raft;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Marshaller interface, for instances that convert objects to {@code byte[]} and back.
+ */
Review Comment:
Please don't suggest making class javadocs single-lined, I won't accept it :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392221333
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/PartitionCommandsMarshaller.java:
##########
@@ -29,7 +29,7 @@ public interface PartitionCommandsMarshaller extends Marshaller {
* Reads required catalog version from the provided buffer.
*
* @param raw Buffer to read from.
- * @return Catalog version.
+ * @return Catalog version. {@code 0} if version is not required for the given command.
Review Comment:
I think it must be -1 and I made a mistake
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1386531891
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/ActionRequest.java:
##########
@@ -32,9 +34,15 @@ public interface ActionRequest extends Message {
String groupId();
/**
- * @return Action's command.
+ * @return Serialized action's command. Specific serialization format may differ from group to group.
+ */
+ byte[] command();
Review Comment:
My answer to myself - no, we shouldn't
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392216311
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftManager.java:
##########
@@ -148,6 +150,7 @@ <T extends RaftGroupService> CompletableFuture<T> startRaftGroupNodeAndWaitNodeR
* @return Future that will be completed with an instance of a Raft group service.
* @throws NodeStoppingException If node stopping intention was detected.
*/
+ @TestOnly
Review Comment:
I was wrong.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392221764
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/ActionRequestProcessor.java:
##########
@@ -82,32 +83,80 @@ public void handleRequest(RpcContext rpcCtx, ActionRequest request) {
return;
}
- JraftServerImpl.DelegatingStateMachine fsm = (JraftServerImpl.DelegatingStateMachine) node.getOptions().getFsm();
+ Marshaller commandsMarshaller = node.getOptions().getCommandsMarshaller();
+
+ assert commandsMarshaller != null : "Marshaller for group " + request.groupId() + " is not found.";
+
+ handleRequestInternal(rpcCtx, node, request, commandsMarshaller);
+ }
+
+ /**
+ * Internal part of the {@link #handleRequest(RpcContext, ActionRequest)}, that contains resolved RAFT node, as well as a commands
+ * marshaller instance. May be conveniently reused in subclasses.
+ */
+ protected void handleRequestInternal(RpcContext rpcCtx, Node node, ActionRequest request, Marshaller commandsMarshaller) {
+ DelegatingStateMachine fsm = (DelegatingStateMachine) node.getOptions().getFsm();
+ RaftGroupListener listener = fsm.getListener();
if (request instanceof WriteActionRequest) {
+ WriteActionRequest writeRequest = (WriteActionRequest)request;
+
+ WriteCommand command = writeRequest.deserializedCommand();
+
+ if (command == null) {
+ command = commandsMarshaller.unmarshall(writeRequest.command());
+ }
+
if (fsm.getListener() instanceof BeforeApplyHandler) {
synchronized (groupIdSyncMonitor(request.groupId())) {
- callOnBeforeApply(request, fsm);
- applyWrite(node, (WriteActionRequest) request, rpcCtx);
+ writeRequest = patchCommandBeforeApply(writeRequest, (BeforeApplyHandler) listener, command, commandsMarshaller);
+
+ applyWrite(node, writeRequest, command, rpcCtx);
}
} else {
- applyWrite(node, (WriteActionRequest) request, rpcCtx);
+ applyWrite(node, writeRequest, command, rpcCtx);
}
} else {
- if (fsm.getListener() instanceof BeforeApplyHandler) {
- callOnBeforeApply(request, fsm);
+ ReadActionRequest readRequest = (ReadActionRequest) request;
+
+ if (listener instanceof BeforeApplyHandler) {
+ ReadCommand command = readRequest.command();
+
+ readRequest = patchCommandBeforeApply(readRequest, (BeforeApplyHandler) listener, command, commandsMarshaller);
}
- applyRead(node, (ReadActionRequest) request, rpcCtx);
+ applyRead(node, readRequest, rpcCtx);
}
}
- private static void callOnBeforeApply(ActionRequest request, DelegatingStateMachine fsm) {
- Command command = request instanceof WriteActionRequest
- ? ((WriteActionRequest) request).command()
- : ((ReadActionRequest) request).command();
-
- ((BeforeApplyHandler) fsm.getListener()).onBeforeApply(command);
+ /**
+ * This method calls {@link BeforeApplyHandler#onBeforeApply(Command)} and returns action request with a serialized version of the
+ * updated command, if it has been updated. Otherwise, the method returns the original {@code request} instance. The reason for such
+ * behavior is the fact that we use {@code byte[]} in action requests, thus modified command should be serialized twice.
+ */
+ private <AR extends ActionRequest> AR patchCommandBeforeApply(
+ AR request,
+ BeforeApplyHandler beforeApplyHandler,
+ Command command,
+ Marshaller commandsMarshaller
+ ) {
+ if (beforeApplyHandler.onBeforeApply(command)) {
+ if (request instanceof WriteActionRequest) {
+ return (AR) factory.writeActionRequest()
+ .groupId(request.groupId())
+ .command(commandsMarshaller.marshall(command))
+ .deserializedCommand((WriteCommand)command)
Review Comment:
Ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392223890
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/CatalogVersionAware.java:
##########
@@ -25,4 +25,9 @@ public interface CatalogVersionAware {
* Returns version that the Catalog must have locally for the node to be allowed to accept this command via replication.
*/
int requiredCatalogVersion();
+
+ /**
+ * Setter for the required catalog version. Should be used with caution. Only object's creator should call it.
+ */
+ void requiredCatalogVersion(int version);
Review Comment:
I change my opinion, this is the right place for this method. PartitionCommand might not be the only sub-interface of CatalogCersionAware in the future
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392222766
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/PartitionCommand.java:
##########
@@ -38,5 +41,15 @@ public interface PartitionCommand extends SafeTimePropagatingCommand, CatalogVer
* Returns version that the Catalog must have locally for the node to be allowed to accept this command via replication.
*/
@Override
+ @Transient
+ @WithSetter
int requiredCatalogVersion();
+
+ /**
+ * Setter for {@link #requiredCatalogVersion()}. Called by the creator or the {@link PartitionCommandsMarshaller} while deserializing.
+ */
+ @Override
+ default void requiredCatalogVersion(int version) {
Review Comment:
Yep
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392211035
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/RaftGroupServiceImpl.java:
##########
@@ -123,6 +125,7 @@ public class RaftGroupServiceImpl implements RaftGroupService {
* @param membersConfiguration Raft members configuration.
* @param leader Group leader.
* @param executor Executor for retrying requests.
+ * @param commandsMarshaller Marshaller that should be used to [de]serialize commands.
Review Comment:
Hm? I'll change the wording if brackets confuse you
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392222766
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/PartitionCommand.java:
##########
@@ -38,5 +41,15 @@ public interface PartitionCommand extends SafeTimePropagatingCommand, CatalogVer
* Returns version that the Catalog must have locally for the node to be allowed to accept this command via replication.
*/
@Override
+ @Transient
+ @WithSetter
int requiredCatalogVersion();
+
+ /**
+ * Setter for {@link #requiredCatalogVersion()}. Called by the creator or the {@link PartitionCommandsMarshaller} while deserializing.
+ */
+ @Override
+ default void requiredCatalogVersion(int version) {
Review Comment:
Yes, but I see that you will try to correct the comment above, perhaps the question will disappear then.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392223890
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/CatalogVersionAware.java:
##########
@@ -25,4 +25,9 @@ public interface CatalogVersionAware {
* Returns version that the Catalog must have locally for the node to be allowed to accept this command via replication.
*/
int requiredCatalogVersion();
+
+ /**
+ * Setter for the required catalog version. Should be used with caution. Only object's creator should call it.
+ */
+ void requiredCatalogVersion(int version);
Review Comment:
I change my opinion, this is the right place for this method. PartitionCommand might not be the only sub-interface of CatalogVersionAware in the future
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1390952999
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/JDKMarshaller.java:
##########
@@ -21,11 +21,14 @@
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.nio.ByteBuffer;
+import org.apache.ignite.internal.raft.Marshaller;
/**
*
*/
public class JDKMarshaller implements Marshaller {
+ public static final Marshaller INSTANCE = new JDKMarshaller();
Review Comment:
Missing javadoc.
##########
modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/server/raft/ItMetaStorageRaftGroupTest.java:
##########
@@ -403,7 +410,8 @@ private List<Pair<RaftServer, RaftGroupService>> prepareJraftMetaStorages(Atomic
raftConfiguration,
membersConfiguration,
true,
- executor
+ executor,
+ commandsMarshaller
).get();
Review Comment:
same
##########
modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageDeserializerGenerator.java:
##########
@@ -118,7 +120,10 @@ private MethodSpec readMessageMethod(MessageClass message, FieldSpec msgField) {
.addParameter(MessageReader.class, "reader")
.addException(MessageMappingException.class);
- List<ExecutableElement> getters = message.getters();
+ List<ExecutableElement> getters = message.getters()
+ .stream()
+ .filter(e -> e.getAnnotation(Transient.class) == null)
+ .collect(Collectors.toList());
Review Comment:
Suggetion: `.collect(toList());`
##########
modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/server/raft/ItMetaStorageRaftGroupTest.java:
##########
@@ -393,7 +399,8 @@ private List<Pair<RaftServer, RaftGroupService>> prepareJraftMetaStorages(Atomic
raftConfiguration,
membersConfiguration,
true,
- executor
+ executor,
+ commandsMarshaller
).get();
Review Comment:
Lets use assertThat(future, willCompleteSuccsessfully());
future.join();
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftManager.java:
##########
@@ -148,6 +150,7 @@ <T extends RaftGroupService> CompletableFuture<T> startRaftGroupNodeAndWaitNodeR
* @return Future that will be completed with an instance of a Raft group service.
* @throws NodeStoppingException If node stopping intention was detected.
*/
+ @TestOnly
Review Comment:
I see that it is used not only in tests but also, for example, in `PlacementDriverManager`.
##########
modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/PlacementDriverManager.java:
##########
@@ -150,7 +150,8 @@ public void start() {
return raftManager.startRaftGroupService(
replicationGroupId,
PeersAndLearners.fromConsistentIds(placementDriverNodes),
- topologyAwareRaftGroupServiceFactory
+ topologyAwareRaftGroupServiceFactory,
+ null
Review Comment:
why `null`?
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/RaftGroupServiceImpl.java:
##########
@@ -452,9 +459,12 @@ public <R> CompletableFuture<R> run(Command cmd) {
Function<Peer, ActionRequest> requestFactory;
if (cmd instanceof WriteCommand) {
+ byte[] commandBytes = commandsMarshaller.marshall(cmd);
+
requestFactory = targetPeer -> factory.writeActionRequest()
.groupId(groupId)
- .command((WriteCommand) cmd)
+ .command(commandBytes)
+ .deserializedCommand((WriteCommand) cmd)
Review Comment:
Please give more description here, why pass the command in raw form.
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/RaftGroupServiceImpl.java:
##########
@@ -123,6 +125,7 @@ public class RaftGroupServiceImpl implements RaftGroupService {
* @param membersConfiguration Raft members configuration.
* @param leader Group leader.
* @param executor Executor for retrying requests.
+ * @param commandsMarshaller Marshaller that should be used to [de]serialize commands.
Review Comment:
```suggestion
* @param commandsMarshaller Marshaller that should be used to de]serialize commands.
```
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/Marshaller.java:
##########
@@ -0,0 +1,53 @@
+/*
+ * 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.raft;
+
+import java.nio.ByteBuffer;
+
+/**
+ * Marshaller interface, for instances that convert objects to {@code byte[]} and back.
+ */
Review Comment:
```suggestion
/** Marshaller interface, for instances that convert objects to {@code byte[]} and back. */
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/PartitionCommand.java:
##########
@@ -38,5 +41,15 @@ public interface PartitionCommand extends SafeTimePropagatingCommand, CatalogVer
* Returns version that the Catalog must have locally for the node to be allowed to accept this command via replication.
*/
@Override
+ @Transient
+ @WithSetter
int requiredCatalogVersion();
+
+ /**
+ * Setter for {@link #requiredCatalogVersion()}. Called by the creator or the {@link PartitionCommandsMarshaller} while deserializing.
+ */
+ @Override
+ default void requiredCatalogVersion(int version) {
Review Comment:
Little strange
##########
modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageSerializerGenerator.java:
##########
@@ -87,7 +89,10 @@ private MethodSpec writeMessageMethod(MessageClass message) {
method.addStatement("$T message = ($T) msg", message.implClassName(), message.implClassName()).addCode("\n");
- List<ExecutableElement> getters = message.getters();
+ List<ExecutableElement> getters = message.getters()
+ .stream()
Review Comment:
```suggestion
List<ExecutableElement> getters = message.getters().stream()
```
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftServiceFactory.java:
##########
@@ -40,6 +40,7 @@ CompletableFuture<T> startRaftGroupService(
ReplicationGroupId groupId,
PeersAndLearners peersAndLearners,
RaftConfiguration raftConfiguration,
- ScheduledExecutorService raftClientExecutor
+ ScheduledExecutorService raftClientExecutor,
+ Marshaller commandsMarshaller
Review Comment:
Missing field description
##########
modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageDeserializerGenerator.java:
##########
@@ -118,7 +120,10 @@ private MethodSpec readMessageMethod(MessageClass message, FieldSpec msgField) {
.addParameter(MessageReader.class, "reader")
.addException(MessageMappingException.class);
- List<ExecutableElement> getters = message.getters();
+ List<ExecutableElement> getters = message.getters()
+ .stream()
Review Comment:
```suggestion
List<ExecutableElement> getters = message.getters().stream()
```
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/service/BeforeApplyHandler.java:
##########
@@ -38,6 +38,7 @@ public interface BeforeApplyHandler {
* this is a place to do it.
*
* @param command The command.
+ * @return {@code true} if command has been changed, {@code false} otherwise.
Review Comment:
What does command changed mean? from a write command to a read command and vice versa?
##########
modules/metastorage/src/integrationTest/java/org/apache/ignite/internal/metastorage/server/raft/ItMetaStorageRaftGroupTest.java:
##########
@@ -413,7 +421,8 @@ private List<Pair<RaftServer, RaftGroupService>> prepareJraftMetaStorages(Atomic
raftConfiguration,
membersConfiguration,
true,
- executor
+ executor,
+ commandsMarshaller
).get();
Review Comment:
same
##########
modules/network-annotation-processor/src/main/java/org/apache/ignite/internal/network/processor/serialization/MessageSerializerGenerator.java:
##########
@@ -87,7 +89,10 @@ private MethodSpec writeMessageMethod(MessageClass message) {
method.addStatement("$T message = ($T) msg", message.implClassName(), message.implClassName()).addCode("\n");
- List<ExecutableElement> getters = message.getters();
+ List<ExecutableElement> getters = message.getters()
+ .stream()
+ .filter(e -> e.getAnnotation(Transient.class) == null)
+ .collect(Collectors.toList());
Review Comment:
Suggession: `.collect(toList());`
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftManager.java:
##########
@@ -165,6 +168,7 @@ CompletableFuture<RaftGroupService> startRaftGroupService(
<T extends RaftGroupService> CompletableFuture<T> startRaftGroupService(
ReplicationGroupId groupId,
PeersAndLearners configuration,
- RaftServiceFactory<T> factory
+ RaftServiceFactory<T> factory,
+ @Nullable Marshaller commandsMarshaller
Review Comment:
Missing field description
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/WriteActionRequest.java:
##########
@@ -19,15 +19,23 @@
import org.apache.ignite.internal.raft.WriteCommand;
import org.apache.ignite.network.annotations.Transferable;
+import org.apache.ignite.network.annotations.Transient;
import org.apache.ignite.raft.jraft.RaftMessageGroup.RpcActionMessageGroup;
+import org.jetbrains.annotations.Nullable;
/**
* Submit a write action to a replication group.
*/
@Transferable(RpcActionMessageGroup.WRITE_ACTION_REQUEST)
public interface WriteActionRequest extends ActionRequest {
/**
- * Returns an action's command.
+ * @return Serialized action's command. Specific serialization format may differ from group to group.
*/
- WriteCommand command();
+ byte[] command();
+
+ /**
+ * @return Original non-serialized command, if available. {@code null} if not.
Review Comment:
Please expand the description and describe why this field is and why it should be used and when it will be `null`.
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/ActionRequestProcessor.java:
##########
@@ -82,32 +83,80 @@ public void handleRequest(RpcContext rpcCtx, ActionRequest request) {
return;
}
- JraftServerImpl.DelegatingStateMachine fsm = (JraftServerImpl.DelegatingStateMachine) node.getOptions().getFsm();
+ Marshaller commandsMarshaller = node.getOptions().getCommandsMarshaller();
+
+ assert commandsMarshaller != null : "Marshaller for group " + request.groupId() + " is not found.";
+
+ handleRequestInternal(rpcCtx, node, request, commandsMarshaller);
+ }
+
+ /**
+ * Internal part of the {@link #handleRequest(RpcContext, ActionRequest)}, that contains resolved RAFT node, as well as a commands
+ * marshaller instance. May be conveniently reused in subclasses.
+ */
+ protected void handleRequestInternal(RpcContext rpcCtx, Node node, ActionRequest request, Marshaller commandsMarshaller) {
+ DelegatingStateMachine fsm = (DelegatingStateMachine) node.getOptions().getFsm();
+ RaftGroupListener listener = fsm.getListener();
if (request instanceof WriteActionRequest) {
+ WriteActionRequest writeRequest = (WriteActionRequest)request;
+
+ WriteCommand command = writeRequest.deserializedCommand();
+
+ if (command == null) {
+ command = commandsMarshaller.unmarshall(writeRequest.command());
+ }
+
if (fsm.getListener() instanceof BeforeApplyHandler) {
synchronized (groupIdSyncMonitor(request.groupId())) {
- callOnBeforeApply(request, fsm);
- applyWrite(node, (WriteActionRequest) request, rpcCtx);
+ writeRequest = patchCommandBeforeApply(writeRequest, (BeforeApplyHandler) listener, command, commandsMarshaller);
+
+ applyWrite(node, writeRequest, command, rpcCtx);
}
} else {
- applyWrite(node, (WriteActionRequest) request, rpcCtx);
+ applyWrite(node, writeRequest, command, rpcCtx);
}
} else {
- if (fsm.getListener() instanceof BeforeApplyHandler) {
- callOnBeforeApply(request, fsm);
+ ReadActionRequest readRequest = (ReadActionRequest) request;
+
+ if (listener instanceof BeforeApplyHandler) {
+ ReadCommand command = readRequest.command();
+
+ readRequest = patchCommandBeforeApply(readRequest, (BeforeApplyHandler) listener, command, commandsMarshaller);
}
- applyRead(node, (ReadActionRequest) request, rpcCtx);
+ applyRead(node, readRequest, rpcCtx);
}
}
- private static void callOnBeforeApply(ActionRequest request, DelegatingStateMachine fsm) {
- Command command = request instanceof WriteActionRequest
- ? ((WriteActionRequest) request).command()
- : ((ReadActionRequest) request).command();
-
- ((BeforeApplyHandler) fsm.getListener()).onBeforeApply(command);
+ /**
+ * This method calls {@link BeforeApplyHandler#onBeforeApply(Command)} and returns action request with a serialized version of the
+ * updated command, if it has been updated. Otherwise, the method returns the original {@code request} instance. The reason for such
+ * behavior is the fact that we use {@code byte[]} in action requests, thus modified command should be serialized twice.
+ */
+ private <AR extends ActionRequest> AR patchCommandBeforeApply(
+ AR request,
+ BeforeApplyHandler beforeApplyHandler,
+ Command command,
+ Marshaller commandsMarshaller
+ ) {
+ if (beforeApplyHandler.onBeforeApply(command)) {
+ if (request instanceof WriteActionRequest) {
+ return (AR) factory.writeActionRequest()
+ .groupId(request.groupId())
+ .command(commandsMarshaller.marshall(command))
+ .deserializedCommand((WriteCommand)command)
Review Comment:
Do you think you shouldn’t do this with reading commands?
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/ActionRequestProcessor.java:
##########
@@ -82,32 +83,80 @@ public void handleRequest(RpcContext rpcCtx, ActionRequest request) {
return;
}
- JraftServerImpl.DelegatingStateMachine fsm = (JraftServerImpl.DelegatingStateMachine) node.getOptions().getFsm();
+ Marshaller commandsMarshaller = node.getOptions().getCommandsMarshaller();
+
+ assert commandsMarshaller != null : "Marshaller for group " + request.groupId() + " is not found.";
+
+ handleRequestInternal(rpcCtx, node, request, commandsMarshaller);
+ }
+
+ /**
+ * Internal part of the {@link #handleRequest(RpcContext, ActionRequest)}, that contains resolved RAFT node, as well as a commands
+ * marshaller instance. May be conveniently reused in subclasses.
+ */
+ protected void handleRequestInternal(RpcContext rpcCtx, Node node, ActionRequest request, Marshaller commandsMarshaller) {
+ DelegatingStateMachine fsm = (DelegatingStateMachine) node.getOptions().getFsm();
+ RaftGroupListener listener = fsm.getListener();
if (request instanceof WriteActionRequest) {
+ WriteActionRequest writeRequest = (WriteActionRequest)request;
+
+ WriteCommand command = writeRequest.deserializedCommand();
+
+ if (command == null) {
+ command = commandsMarshaller.unmarshall(writeRequest.command());
+ }
+
if (fsm.getListener() instanceof BeforeApplyHandler) {
synchronized (groupIdSyncMonitor(request.groupId())) {
- callOnBeforeApply(request, fsm);
- applyWrite(node, (WriteActionRequest) request, rpcCtx);
+ writeRequest = patchCommandBeforeApply(writeRequest, (BeforeApplyHandler) listener, command, commandsMarshaller);
+
+ applyWrite(node, writeRequest, command, rpcCtx);
}
} else {
- applyWrite(node, (WriteActionRequest) request, rpcCtx);
+ applyWrite(node, writeRequest, command, rpcCtx);
}
} else {
- if (fsm.getListener() instanceof BeforeApplyHandler) {
- callOnBeforeApply(request, fsm);
+ ReadActionRequest readRequest = (ReadActionRequest) request;
+
+ if (listener instanceof BeforeApplyHandler) {
+ ReadCommand command = readRequest.command();
+
+ readRequest = patchCommandBeforeApply(readRequest, (BeforeApplyHandler) listener, command, commandsMarshaller);
}
- applyRead(node, (ReadActionRequest) request, rpcCtx);
+ applyRead(node, readRequest, rpcCtx);
}
}
- private static void callOnBeforeApply(ActionRequest request, DelegatingStateMachine fsm) {
- Command command = request instanceof WriteActionRequest
- ? ((WriteActionRequest) request).command()
- : ((ReadActionRequest) request).command();
-
- ((BeforeApplyHandler) fsm.getListener()).onBeforeApply(command);
+ /**
+ * This method calls {@link BeforeApplyHandler#onBeforeApply(Command)} and returns action request with a serialized version of the
+ * updated command, if it has been updated. Otherwise, the method returns the original {@code request} instance. The reason for such
+ * behavior is the fact that we use {@code byte[]} in action requests, thus modified command should be serialized twice.
+ */
+ private <AR extends ActionRequest> AR patchCommandBeforeApply(
+ AR request,
+ BeforeApplyHandler beforeApplyHandler,
+ Command command,
+ Marshaller commandsMarshaller
+ ) {
+ if (beforeApplyHandler.onBeforeApply(command)) {
Review Comment:
maybe:
`if (!beforeApplyHandler.onBeforeApply(command)) {
return request.
}
... other code`
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/ActionRequestProcessor.java:
##########
@@ -82,32 +83,80 @@ public void handleRequest(RpcContext rpcCtx, ActionRequest request) {
return;
}
- JraftServerImpl.DelegatingStateMachine fsm = (JraftServerImpl.DelegatingStateMachine) node.getOptions().getFsm();
+ Marshaller commandsMarshaller = node.getOptions().getCommandsMarshaller();
+
+ assert commandsMarshaller != null : "Marshaller for group " + request.groupId() + " is not found.";
+
+ handleRequestInternal(rpcCtx, node, request, commandsMarshaller);
+ }
+
+ /**
+ * Internal part of the {@link #handleRequest(RpcContext, ActionRequest)}, that contains resolved RAFT node, as well as a commands
+ * marshaller instance. May be conveniently reused in subclasses.
+ */
+ protected void handleRequestInternal(RpcContext rpcCtx, Node node, ActionRequest request, Marshaller commandsMarshaller) {
+ DelegatingStateMachine fsm = (DelegatingStateMachine) node.getOptions().getFsm();
+ RaftGroupListener listener = fsm.getListener();
if (request instanceof WriteActionRequest) {
+ WriteActionRequest writeRequest = (WriteActionRequest)request;
+
+ WriteCommand command = writeRequest.deserializedCommand();
+
+ if (command == null) {
+ command = commandsMarshaller.unmarshall(writeRequest.command());
+ }
+
if (fsm.getListener() instanceof BeforeApplyHandler) {
synchronized (groupIdSyncMonitor(request.groupId())) {
- callOnBeforeApply(request, fsm);
- applyWrite(node, (WriteActionRequest) request, rpcCtx);
+ writeRequest = patchCommandBeforeApply(writeRequest, (BeforeApplyHandler) listener, command, commandsMarshaller);
+
+ applyWrite(node, writeRequest, command, rpcCtx);
}
} else {
- applyWrite(node, (WriteActionRequest) request, rpcCtx);
+ applyWrite(node, writeRequest, command, rpcCtx);
}
} else {
- if (fsm.getListener() instanceof BeforeApplyHandler) {
- callOnBeforeApply(request, fsm);
+ ReadActionRequest readRequest = (ReadActionRequest) request;
+
+ if (listener instanceof BeforeApplyHandler) {
+ ReadCommand command = readRequest.command();
+
+ readRequest = patchCommandBeforeApply(readRequest, (BeforeApplyHandler) listener, command, commandsMarshaller);
}
- applyRead(node, (ReadActionRequest) request, rpcCtx);
+ applyRead(node, readRequest, rpcCtx);
}
}
- private static void callOnBeforeApply(ActionRequest request, DelegatingStateMachine fsm) {
- Command command = request instanceof WriteActionRequest
- ? ((WriteActionRequest) request).command()
- : ((ReadActionRequest) request).command();
-
- ((BeforeApplyHandler) fsm.getListener()).onBeforeApply(command);
+ /**
+ * This method calls {@link BeforeApplyHandler#onBeforeApply(Command)} and returns action request with a serialized version of the
+ * updated command, if it has been updated. Otherwise, the method returns the original {@code request} instance. The reason for such
+ * behavior is the fact that we use {@code byte[]} in action requests, thus modified command should be serialized twice.
+ */
+ private <AR extends ActionRequest> AR patchCommandBeforeApply(
+ AR request,
+ BeforeApplyHandler beforeApplyHandler,
+ Command command,
+ Marshaller commandsMarshaller
+ ) {
+ if (beforeApplyHandler.onBeforeApply(command)) {
+ if (request instanceof WriteActionRequest) {
+ return (AR) factory.writeActionRequest()
+ .groupId(request.groupId())
+ .command(commandsMarshaller.marshall(command))
+ .deserializedCommand((WriteCommand)command)
Review Comment:
Do you think you shouldn’t do this with reading commands?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/CatalogVersionAware.java:
##########
@@ -25,4 +25,9 @@ public interface CatalogVersionAware {
* Returns version that the Catalog must have locally for the node to be allowed to accept this command via replication.
*/
int requiredCatalogVersion();
+
+ /**
+ * Setter for the required catalog version. Should be used with caution. Only object's creator should call it.
+ */
+ void requiredCatalogVersion(int version);
Review Comment:
Shouldn't `org.apache.ignite.internal.table.distributed.command.PartitionCommand#requiredCatalogVersion()` hide this setter?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/PartitionCommandsMarshaller.java:
##########
@@ -29,7 +29,7 @@ public interface PartitionCommandsMarshaller extends Marshaller {
* Reads required catalog version from the provided buffer.
*
* @param raw Buffer to read from.
- * @return Catalog version.
+ * @return Catalog version. {@code 0} if version is not required for the given command.
Review Comment:
`-1` is more obvious.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov merged PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392217544
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/ActionRequestProcessor.java:
##########
@@ -82,32 +83,80 @@ public void handleRequest(RpcContext rpcCtx, ActionRequest request) {
return;
}
- JraftServerImpl.DelegatingStateMachine fsm = (JraftServerImpl.DelegatingStateMachine) node.getOptions().getFsm();
+ Marshaller commandsMarshaller = node.getOptions().getCommandsMarshaller();
+
+ assert commandsMarshaller != null : "Marshaller for group " + request.groupId() + " is not found.";
+
+ handleRequestInternal(rpcCtx, node, request, commandsMarshaller);
+ }
+
+ /**
+ * Internal part of the {@link #handleRequest(RpcContext, ActionRequest)}, that contains resolved RAFT node, as well as a commands
+ * marshaller instance. May be conveniently reused in subclasses.
+ */
+ protected void handleRequestInternal(RpcContext rpcCtx, Node node, ActionRequest request, Marshaller commandsMarshaller) {
+ DelegatingStateMachine fsm = (DelegatingStateMachine) node.getOptions().getFsm();
+ RaftGroupListener listener = fsm.getListener();
if (request instanceof WriteActionRequest) {
+ WriteActionRequest writeRequest = (WriteActionRequest)request;
+
+ WriteCommand command = writeRequest.deserializedCommand();
+
+ if (command == null) {
+ command = commandsMarshaller.unmarshall(writeRequest.command());
+ }
+
if (fsm.getListener() instanceof BeforeApplyHandler) {
synchronized (groupIdSyncMonitor(request.groupId())) {
- callOnBeforeApply(request, fsm);
- applyWrite(node, (WriteActionRequest) request, rpcCtx);
+ writeRequest = patchCommandBeforeApply(writeRequest, (BeforeApplyHandler) listener, command, commandsMarshaller);
+
+ applyWrite(node, writeRequest, command, rpcCtx);
}
} else {
- applyWrite(node, (WriteActionRequest) request, rpcCtx);
+ applyWrite(node, writeRequest, command, rpcCtx);
}
} else {
- if (fsm.getListener() instanceof BeforeApplyHandler) {
- callOnBeforeApply(request, fsm);
+ ReadActionRequest readRequest = (ReadActionRequest) request;
+
+ if (listener instanceof BeforeApplyHandler) {
+ ReadCommand command = readRequest.command();
+
+ readRequest = patchCommandBeforeApply(readRequest, (BeforeApplyHandler) listener, command, commandsMarshaller);
}
- applyRead(node, (ReadActionRequest) request, rpcCtx);
+ applyRead(node, readRequest, rpcCtx);
}
}
- private static void callOnBeforeApply(ActionRequest request, DelegatingStateMachine fsm) {
- Command command = request instanceof WriteActionRequest
- ? ((WriteActionRequest) request).command()
- : ((ReadActionRequest) request).command();
-
- ((BeforeApplyHandler) fsm.getListener()).onBeforeApply(command);
+ /**
+ * This method calls {@link BeforeApplyHandler#onBeforeApply(Command)} and returns action request with a serialized version of the
+ * updated command, if it has been updated. Otherwise, the method returns the original {@code request} instance. The reason for such
+ * behavior is the fact that we use {@code byte[]} in action requests, thus modified command should be serialized twice.
+ */
+ private <AR extends ActionRequest> AR patchCommandBeforeApply(
+ AR request,
+ BeforeApplyHandler beforeApplyHandler,
+ Command command,
+ Marshaller commandsMarshaller
+ ) {
+ if (beforeApplyHandler.onBeforeApply(command)) {
+ if (request instanceof WriteActionRequest) {
+ return (AR) factory.writeActionRequest()
+ .groupId(request.groupId())
+ .command(commandsMarshaller.marshall(command))
+ .deserializedCommand((WriteCommand)command)
Review Comment:
Yes, that's why I didn't do it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392216912
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/ActionRequestProcessor.java:
##########
@@ -82,32 +83,80 @@ public void handleRequest(RpcContext rpcCtx, ActionRequest request) {
return;
}
- JraftServerImpl.DelegatingStateMachine fsm = (JraftServerImpl.DelegatingStateMachine) node.getOptions().getFsm();
+ Marshaller commandsMarshaller = node.getOptions().getCommandsMarshaller();
+
+ assert commandsMarshaller != null : "Marshaller for group " + request.groupId() + " is not found.";
+
+ handleRequestInternal(rpcCtx, node, request, commandsMarshaller);
+ }
+
+ /**
+ * Internal part of the {@link #handleRequest(RpcContext, ActionRequest)}, that contains resolved RAFT node, as well as a commands
+ * marshaller instance. May be conveniently reused in subclasses.
+ */
+ protected void handleRequestInternal(RpcContext rpcCtx, Node node, ActionRequest request, Marshaller commandsMarshaller) {
+ DelegatingStateMachine fsm = (DelegatingStateMachine) node.getOptions().getFsm();
+ RaftGroupListener listener = fsm.getListener();
if (request instanceof WriteActionRequest) {
+ WriteActionRequest writeRequest = (WriteActionRequest)request;
+
+ WriteCommand command = writeRequest.deserializedCommand();
+
+ if (command == null) {
+ command = commandsMarshaller.unmarshall(writeRequest.command());
+ }
+
if (fsm.getListener() instanceof BeforeApplyHandler) {
synchronized (groupIdSyncMonitor(request.groupId())) {
- callOnBeforeApply(request, fsm);
- applyWrite(node, (WriteActionRequest) request, rpcCtx);
+ writeRequest = patchCommandBeforeApply(writeRequest, (BeforeApplyHandler) listener, command, commandsMarshaller);
+
+ applyWrite(node, writeRequest, command, rpcCtx);
}
} else {
- applyWrite(node, (WriteActionRequest) request, rpcCtx);
+ applyWrite(node, writeRequest, command, rpcCtx);
}
} else {
- if (fsm.getListener() instanceof BeforeApplyHandler) {
- callOnBeforeApply(request, fsm);
+ ReadActionRequest readRequest = (ReadActionRequest) request;
+
+ if (listener instanceof BeforeApplyHandler) {
+ ReadCommand command = readRequest.command();
+
+ readRequest = patchCommandBeforeApply(readRequest, (BeforeApplyHandler) listener, command, commandsMarshaller);
}
- applyRead(node, (ReadActionRequest) request, rpcCtx);
+ applyRead(node, readRequest, rpcCtx);
}
}
- private static void callOnBeforeApply(ActionRequest request, DelegatingStateMachine fsm) {
- Command command = request instanceof WriteActionRequest
- ? ((WriteActionRequest) request).command()
- : ((ReadActionRequest) request).command();
-
- ((BeforeApplyHandler) fsm.getListener()).onBeforeApply(command);
+ /**
+ * This method calls {@link BeforeApplyHandler#onBeforeApply(Command)} and returns action request with a serialized version of the
+ * updated command, if it has been updated. Otherwise, the method returns the original {@code request} instance. The reason for such
+ * behavior is the fact that we use {@code byte[]} in action requests, thus modified command should be serialized twice.
+ */
+ private <AR extends ActionRequest> AR patchCommandBeforeApply(
+ AR request,
+ BeforeApplyHandler beforeApplyHandler,
+ Command command,
+ Marshaller commandsMarshaller
+ ) {
+ if (beforeApplyHandler.onBeforeApply(command)) {
Review Comment:
Sure
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392211782
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/RaftGroupServiceImpl.java:
##########
@@ -452,9 +459,12 @@ public <R> CompletableFuture<R> run(Command cmd) {
Function<Peer, ActionRequest> requestFactory;
if (cmd instanceof WriteCommand) {
+ byte[] commandBytes = commandsMarshaller.marshall(cmd);
+
requestFactory = targetPeer -> factory.writeActionRequest()
.groupId(groupId)
- .command((WriteCommand) cmd)
+ .command(commandBytes)
+ .deserializedCommand((WriteCommand) cmd)
Review Comment:
Ok
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392199429
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/RaftManager.java:
##########
@@ -148,6 +150,7 @@ <T extends RaftGroupService> CompletableFuture<T> startRaftGroupNodeAndWaitNodeR
* @return Future that will be completed with an instance of a Raft group service.
* @throws NodeStoppingException If node stopping intention was detected.
*/
+ @TestOnly
Review Comment:
Ok, then this change is obsolete, thank you for noticing!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392207437
##########
modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/PlacementDriverManager.java:
##########
@@ -150,7 +150,8 @@ public void start() {
return raftManager.startRaftGroupService(
replicationGroupId,
PeersAndLearners.fromConsistentIds(placementDriverNodes),
- topologyAwareRaftGroupServiceFactory
+ topologyAwareRaftGroupServiceFactory,
+ null
Review Comment:
Added a comment. We use default marshaller here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392288220
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/schema/PartitionCommandsMarshaller.java:
##########
@@ -25,11 +25,16 @@
* representation of the object.
*/
public interface PartitionCommandsMarshaller extends Marshaller {
+ /**
+ * Used instead of a required catalog version when there is no requirement.
+ */
+ int NO_VERSION_REQUIRED = -1;
+
/**
* Reads required catalog version from the provided buffer.
*
* @param raw Buffer to read from.
- * @return Catalog version. {@code 0} if version is not required for the given command.
+ * @return Catalog version. {@code -1} if version is not required for the given command.
Review Comment:
U cau use {@value #NO_VERSION_REQUIRED}
##########
modules/network-api/src/main/java/org/apache/ignite/network/annotations/Transient.java:
##########
@@ -24,7 +24,7 @@
/**
* Annotation for methods of {@link Transferable} interface, that makes generated serializer/deserializer ignore the property.
- */
+ *///TODO DEVNOTES
Review Comment:
I think I forgot to indicate the ticket or something like that
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/CatalogVersionAware.java:
##########
@@ -25,4 +25,9 @@ public interface CatalogVersionAware {
* Returns version that the Catalog must have locally for the node to be allowed to accept this command via replication.
*/
int requiredCatalogVersion();
+
+ /**
+ * Setter for the required catalog version. Should be used with caution. Only object's creator should call it.
+ */
+ void requiredCatalogVersion(int version);
Review Comment:
Ok
##########
modules/raft/src/main/java/org/apache/ignite/internal/raft/RaftGroupServiceImpl.java:
##########
@@ -452,9 +459,12 @@ public <R> CompletableFuture<R> run(Command cmd) {
Function<Peer, ActionRequest> requestFactory;
if (cmd instanceof WriteCommand) {
+ byte[] commandBytes = commandsMarshaller.marshall(cmd);
+
requestFactory = targetPeer -> factory.writeActionRequest()
.groupId(groupId)
- .command((WriteCommand) cmd)
+ .command(commandBytes)
+ .deserializedCommand((WriteCommand) cmd)
Review Comment:
Maybe we can add more? for example, "so that the local node does not desiralize the command several times".
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392220303
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/PartitionCommand.java:
##########
@@ -38,5 +41,15 @@ public interface PartitionCommand extends SafeTimePropagatingCommand, CatalogVer
* Returns version that the Catalog must have locally for the node to be allowed to accept this command via replication.
*/
@Override
+ @Transient
+ @WithSetter
int requiredCatalogVersion();
+
+ /**
+ * Setter for {@link #requiredCatalogVersion()}. Called by the creator or the {@link PartitionCommandsMarshaller} while deserializing.
+ */
+ @Override
+ default void requiredCatalogVersion(int version) {
Review Comment:
What exactly is strange? The fact that javadoc differs from the parent class?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "ibessonov (via GitHub)" <gi...@apache.org>.
ibessonov commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392219015
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/command/CatalogVersionAware.java:
##########
@@ -25,4 +25,9 @@ public interface CatalogVersionAware {
* Returns version that the Catalog must have locally for the node to be allowed to accept this command via replication.
*/
int requiredCatalogVersion();
+
+ /**
+ * Setter for the required catalog version. Should be used with caution. Only object's creator should call it.
+ */
+ void requiredCatalogVersion(int version);
Review Comment:
You mean moving it to subclass? Yes, I think I can do it, that's probably a good advice
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] IGNITE-19453 Move raft command serialization before the action request creation. [ignite-3]
Posted by "tkalkirill (via GitHub)" <gi...@apache.org>.
tkalkirill commented on code in PR #2772:
URL: https://github.com/apache/ignite-3/pull/2772#discussion_r1392213384
##########
modules/raft-api/src/main/java/org/apache/ignite/internal/raft/service/BeforeApplyHandler.java:
##########
@@ -38,6 +38,7 @@ public interface BeforeApplyHandler {
* this is a place to do it.
*
* @param command The command.
+ * @return {@code true} if command has been changed, {@code false} otherwise.
Review Comment:
Please reflect this in the documentation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: notifications-unsubscribe@ignite.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org