You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "sanpwc (via GitHub)" <gi...@apache.org> on 2023/11/08 20:38:21 UTC
[PR] IGNITE-20709 [ignite-3]
sanpwc opened a new pull request, #2820:
URL: https://github.com/apache/ignite-3/pull/2820
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-20709 [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389422294
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1587,34 +1594,53 @@ private CompletableFuture<Object> finishTransaction(
HybridTimestamp tsForCatalogVersion = commit ? commitTimestamp : hybridClock.now();
return reliableCatalogVersionFor(tsForCatalogVersion)
- .thenCompose(catalogVersion -> {
- synchronized (commandProcessingLinearizationMutex) {
- FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
- .txId(txId)
- .commit(commit)
- .safeTimeLong(hybridClock.nowLong())
- .txCoordinatorId(txCoordinatorId)
- .requiredCatalogVersion(catalogVersion)
- .tablePartitionIds(
- aggregatedGroupIds.stream()
- .map(PartitionReplicaListener::tablePartitionId)
- .collect(toList())
- );
-
- if (commit) {
- finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
- }
-
- return raftClient.run(finishTxCmdBldr.build());
- }
- })
+ .thenCompose(catalogVersion -> applyFinishCommand(
+ txId,
+ commit,
+ commitTimestamp,
+ txCoordinatorId,
+ catalogVersion,
+ aggregatedGroupIds.stream()
+ .map(PartitionReplicaListener::tablePartitionId)
+ .collect(toList())
+ )
+ )
.whenComplete((o, throwable) -> {
TxState txState = commit ? COMMITED : ABORTED;
markFinished(txId, txState, commitTimestamp);
});
}
+ private CompletableFuture<Object> applyFinishCommand(
+ UUID transactionId,
+ boolean commit,
+ HybridTimestamp commitTimestamp,
+ String txCoordinatorId,
+ int catalogVersion,
+ List<TablePartitionIdMessage> tablePartitionIds
+ ) {
+ // TODO: synchonized should go here. Not within finishTransaction but with all apply command.
Review Comment:
TODO was outdated. Removed.
--
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-20709 [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389476809
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2383,6 +2424,35 @@ private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
});
}
+ private void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<Object> resultFuture) {
+ raftClient.run(cmd).whenComplete((res, ex) -> {
+ if (ex != null) {
+ if (ex instanceof SafeTimeReorderException || ex.getCause() instanceof SafeTimeReorderException) {
+ assert cmd instanceof SafeTimePropagatingCommand;
Review Comment:
Do you mean that applyCmdWithRetryOnSafeTimeReorderException should have SafeTimePropagatingCommand instead of Command as a parameter? I don't think it's convenient because in that case we should skip calling applyCmdWithRetryOnSafeTimeReorderException for all commands except SafeTimePropagatingCommand and skip raftClientRun directly.
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1390747367
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2420,34 +2490,40 @@ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
);
if (!cmd.full()) {
- CompletableFuture<UUID> fut = applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 Temporary code below
+ synchronized (safeTime) {
+ storageUpdateHandler.handleUpdate(
Review Comment:
This explains why for 1PC we do 'send to RAFT first, write to the Primary storage second'. But why for !1PC do we do it in the opposite order?
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1390741098
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2383,6 +2424,35 @@ private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
});
}
+ private void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<Object> resultFuture) {
+ raftClient.run(cmd).whenComplete((res, ex) -> {
+ if (ex != null) {
+ if (ex instanceof SafeTimeReorderException || ex.getCause() instanceof SafeTimeReorderException) {
+ assert cmd instanceof SafeTimePropagatingCommand;
+
+ SafeTimePropagatingCommand safeTimePropagatingCommand = (SafeTimePropagatingCommand) cmd;
+
+ HybridTimestamp safeTimeForRetry = hybridClock.now();
+
+ if ((cmd instanceof UpdateCommand && !((UpdateCommand) cmd).full())
Review Comment:
Thanks for adding the comment, but my confusion is caused by another question: why DON'T we update safe time in case of 1PC?
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1390985251
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -148,7 +159,17 @@ public void onWrite(Iterator<CommandClosure<WriteCommand>> iterator) {
iterator.forEachRemaining((CommandClosure<? extends WriteCommand> clo) -> {
Command command = clo.command();
- // LOG.info("CMD {}", command.getClass().getName());
+ if (command instanceof SafeTimePropagatingCommand) {
+ SafeTimePropagatingCommand cmd = (SafeTimePropagatingCommand) command;
+ long proposedSafeTime = cmd.safeTime().longValue();
+
+ if (proposedSafeTime > maxObservableSafeTimeVerifier) {
+ maxObservableSafeTimeVerifier = proposedSafeTime;
+ } else {
+ assert false : "Safe time reordering detected [current=" + maxObservableSafeTimeVerifier
+ + ", proposed=" + proposedSafeTime + "]";
+ }
Review Comment:
Why do you make the code so complicated? I suspect you should write like this:
```
assert roposedSafeTime <= maxObservableSafeTimeVerifier : "Safe time reordering detected [current=" + maxObservableSafeTimeVerifier + ", proposed=" + proposedSafeTime + "].";
maxObservableSafeTimeVerifier = proposedSafeTime;
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -187,6 +190,9 @@ public class PartitionReplicaListener implements ReplicaListener {
/** Factory for creating replica command messages. */
private static final ReplicaMessagesFactory REPLICA_MESSAGES_FACTORY = new ReplicaMessagesFactory();
+ /** Replication retries limit. */
+ private static final int MAX_RETIES_ON_SAFE_TIME_REORDERING = 1000;
Review Comment:
Why is it meant to limit attempts if this is not a recoverable case?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -148,7 +159,17 @@ public void onWrite(Iterator<CommandClosure<WriteCommand>> iterator) {
iterator.forEachRemaining((CommandClosure<? extends WriteCommand> clo) -> {
Command command = clo.command();
- // LOG.info("CMD {}", command.getClass().getName());
+ if (command instanceof SafeTimePropagatingCommand) {
+ SafeTimePropagatingCommand cmd = (SafeTimePropagatingCommand) command;
+ long proposedSafeTime = cmd.safeTime().longValue();
+
+ if (proposedSafeTime > maxObservableSafeTimeVerifier) {
+ maxObservableSafeTimeVerifier = proposedSafeTime;
+ } else {
+ assert false : "Safe time reordering detected [current=" + maxObservableSafeTimeVerifier
+ + ", proposed=" + proposedSafeTime + "]";
+ }
Review Comment:
If the `maxObservableSafeTimeVerifier` is needed only for verification, it would be better to remove it altogether.
--
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-20709 [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389522573
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2420,34 +2490,40 @@ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
);
if (!cmd.full()) {
- CompletableFuture<UUID> fut = applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 Temporary code below
+ synchronized (safeTime) {
+ storageUpdateHandler.handleUpdate(
Review Comment:
Because in case of full command we apply not the write intent but write itself. Writes doesn't require txn state resolution and thus visible to other transactions directly. All in all that means that we should prevent write visibility in case of replication failure. The reliable way to do it is to put replication prior to primary replica adjustment.
Why we need direct put on primary replica?
Not only because replication is finished when it's finished on majoirty that may miss primary but also because primary replica may not be a part of replication group.
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1390771455
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2383,6 +2424,35 @@ private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
});
}
+ private void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<Object> resultFuture) {
+ raftClient.run(cmd).whenComplete((res, ex) -> {
+ if (ex != null) {
+ if (ex instanceof SafeTimeReorderException || ex.getCause() instanceof SafeTimeReorderException) {
+ assert cmd instanceof SafeTimePropagatingCommand;
+
+ SafeTimePropagatingCommand safeTimePropagatingCommand = (SafeTimePropagatingCommand) cmd;
+
+ HybridTimestamp safeTimeForRetry = hybridClock.now();
+
+ if ((cmd instanceof UpdateCommand && !((UpdateCommand) cmd).full())
Review Comment:
Because in case of 1PC replication goes prior to local storage update.
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1390934167
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2420,34 +2490,40 @@ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
);
if (!cmd.full()) {
- CompletableFuture<UUID> fut = applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 Temporary code below
+ synchronized (safeTime) {
+ storageUpdateHandler.handleUpdate(
+ cmd.txId(),
+ cmd.rowUuid(),
+ cmd.tablePartitionId().asTablePartitionId(),
+ cmd.rowToUpdate(),
+ true,
+ null,
+ null,
+ null);
+
+ updateTrackerIgnoringTrackerClosedException(safeTime, cmd.safeTime());
+ }
+
+ CompletableFuture<Object> resultFuture = new CompletableFuture<>();
+
+ applyCmdWithExceptionHandling(cmd, resultFuture);
Review Comment:
True
--
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-20709 [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389428714
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1677,32 +1703,45 @@ private CompletableFuture<Void> processTxCleanupAction(TxCleanupReplicaRequest r
return reliableCatalogVersionFor(commandTimestamp)
.thenCompose(catalogVersion -> {
- synchronized (commandProcessingLinearizationMutex) {
- TxCleanupCommand txCleanupCmd = MSG_FACTORY.txCleanupCommand()
- .txId(request.txId())
- .commit(request.commit())
- .commitTimestampLong(request.commitTimestampLong())
- .safeTimeLong(hybridClock.nowLong())
- .txCoordinatorId(getTxCoordinatorId(request.txId()))
- .requiredCatalogVersion(catalogVersion)
- .build();
-
- storageUpdateHandler.handleTransactionCleanup(request.txId(), request.commit(), request.commitTimestamp());
-
- raftClient.run(txCleanupCmd)
- .exceptionally(e -> {
- LOG.warn("Failed to complete transaction cleanup command [txId=" + request.txId() + ']', e);
-
- return completedFuture(null);
- });
- }
+ applyCleanupCommand(
+ request.txId(),
+ request.commit(),
+ request.commitTimestamp(),
+ request.commitTimestampLong(),
+ catalogVersion
+ );
return allOffFuturesExceptionIgnored(txReadFutures, request)
.thenRun(() -> releaseTxLocks(request.txId()));
});
});
}
+ private CompletableFuture<Void> applyCleanupCommand(
+ UUID transactionId,
+ boolean commit,
+ HybridTimestamp commitTimestamp,
+ long commitTimestampLong,
+ int catalogVersion
+ ) {
+ TxCleanupCommand txCleanupCmd = MSG_FACTORY.txCleanupCommand()
+ .txId(transactionId)
+ .commit(commit)
+ .commitTimestampLong(commitTimestampLong)
+ .safeTimeLong(hybridClock.nowLong())
+ .txCoordinatorId(getTxCoordinatorId(transactionId))
+ .requiredCatalogVersion(catalogVersion)
+ .build();
+
+ storageUpdateHandler.handleTransactionCleanup(transactionId, commit, commitTimestamp);
+
+ CompletableFuture<Object> resultFuture = new CompletableFuture<>();
+
+ applyCmdWithRetryOnSafeTimeReorderException(txCleanupCmd, resultFuture);
+
+ return resultFuture.thenApply(res -> null);
Review Comment:
No, it's a bug. Fixed.
--
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-20709 [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389420045
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/error/RaftError.java:
##########
@@ -238,7 +238,12 @@ public enum RaftError {
/**
* Permission denied
*/
- EACCES(1016);
+ EACCES(1016),
+
+ /**
+ * Command reordering detected
+ */
+ EREORDER(1017);
Review Comment:
Sure. Fixed.
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1391143774
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -187,6 +190,9 @@ public class PartitionReplicaListener implements ReplicaListener {
/** Factory for creating replica command messages. */
private static final ReplicaMessagesFactory REPLICA_MESSAGES_FACTORY = new ReplicaMessagesFactory();
+ /** Replication retries limit. */
+ private static final int MAX_RETIES_ON_SAFE_TIME_REORDERING = 1000;
Review Comment:
Please check prior Roma's comments.
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1391149721
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -148,7 +159,17 @@ public void onWrite(Iterator<CommandClosure<WriteCommand>> iterator) {
iterator.forEachRemaining((CommandClosure<? extends WriteCommand> clo) -> {
Command command = clo.command();
- // LOG.info("CMD {}", command.getClass().getName());
+ if (command instanceof SafeTimePropagatingCommand) {
+ SafeTimePropagatingCommand cmd = (SafeTimePropagatingCommand) command;
+ long proposedSafeTime = cmd.safeTime().longValue();
+
+ if (proposedSafeTime > maxObservableSafeTimeVerifier) {
+ maxObservableSafeTimeVerifier = proposedSafeTime;
+ } else {
+ assert false : "Safe time reordering detected [current=" + maxObservableSafeTimeVerifier
+ + ", proposed=" + proposedSafeTime + "]";
+ }
Review Comment:
> it would be better to remove it altogether.
I don't think so. Given assertion will ease tests failures analysis. In case of unexpected data we will see whether there was a reordering or not.
> assert roposedSafeTime <= maxObservableSafeTimeVerifier : "Safe time reordering detected [current=" + maxObservableSafeTimeVerifier + ", proposed=" + proposedSafeTime + "].";
maxObservableSafeTimeVerifier = proposedSafeTime;
Agreed. Fixed. Except the part that proposed should be greater than maxObservableSafeTimeVerifier
--
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-20709 [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389031580
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1587,34 +1594,53 @@ private CompletableFuture<Object> finishTransaction(
HybridTimestamp tsForCatalogVersion = commit ? commitTimestamp : hybridClock.now();
return reliableCatalogVersionFor(tsForCatalogVersion)
- .thenCompose(catalogVersion -> {
- synchronized (commandProcessingLinearizationMutex) {
- FinishTxCommandBuilder finishTxCmdBldr = MSG_FACTORY.finishTxCommand()
- .txId(txId)
- .commit(commit)
- .safeTimeLong(hybridClock.nowLong())
- .txCoordinatorId(txCoordinatorId)
- .requiredCatalogVersion(catalogVersion)
- .tablePartitionIds(
- aggregatedGroupIds.stream()
- .map(PartitionReplicaListener::tablePartitionId)
- .collect(toList())
- );
-
- if (commit) {
- finishTxCmdBldr.commitTimestampLong(commitTimestamp.longValue());
- }
-
- return raftClient.run(finishTxCmdBldr.build());
- }
- })
+ .thenCompose(catalogVersion -> applyFinishCommand(
+ txId,
+ commit,
+ commitTimestamp,
+ txCoordinatorId,
+ catalogVersion,
+ aggregatedGroupIds.stream()
+ .map(PartitionReplicaListener::tablePartitionId)
+ .collect(toList())
+ )
+ )
.whenComplete((o, throwable) -> {
TxState txState = commit ? COMMITED : ABORTED;
markFinished(txId, txState, commitTimestamp);
});
}
+ private CompletableFuture<Object> applyFinishCommand(
+ UUID transactionId,
+ boolean commit,
+ HybridTimestamp commitTimestamp,
+ String txCoordinatorId,
+ int catalogVersion,
+ List<TablePartitionIdMessage> tablePartitionIds
+ ) {
+ // TODO: synchonized should go here. Not within finishTransaction but with all apply command.
Review Comment:
Please reference a JIRA issue in the TODO
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -99,6 +101,15 @@ public class PartitionListener implements RaftGroupListener {
/** Storage index tracker. */
private final PendingComparableValuesTracker<Long, Void> storageIndexTracker;
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20826 Restore on restart
+ /** Is used in order to detect and retry safe time reordering within onBeforeApply. */
+ private long maxObservableSafeTime = -1;
+
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20826 Restore on restart
+ /** Is used in order to assert safe time reordering within onWrite. */
+ private long maxObservableSafeTimeVerificator = -1;
Review Comment:
verificator -> verifier?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2383,6 +2424,35 @@ private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
});
}
+ private void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<Object> resultFuture) {
+ raftClient.run(cmd).whenComplete((res, ex) -> {
+ if (ex != null) {
+ if (ex instanceof SafeTimeReorderException || ex.getCause() instanceof SafeTimeReorderException) {
+ assert cmd instanceof SafeTimePropagatingCommand;
+
+ SafeTimePropagatingCommand safeTimePropagatingCommand = (SafeTimePropagatingCommand) cmd;
+
+ HybridTimestamp safeTimeForRetry = hybridClock.now();
+
+ if ((cmd instanceof UpdateCommand && !((UpdateCommand) cmd).full())
+ || (cmd instanceof UpdateAllCommand && !((UpdateAllCommand) cmd).full())) {
+ synchronized (safeTime) {
+ updateTrackerIgnoringTrackerClosedException(safeTime, safeTimeForRetry);
+ }
+ }
+
+ safeTimePropagatingCommand.safeTimeLong(safeTimeForRetry.longValue());
+
+ applyCmdWithRetryOnSafeTimeReorderException(safeTimePropagatingCommand, resultFuture);
Review Comment:
Is it possible to add some kind of a safeguard to check/assert that we are not retrying infinitely due to some bug?
##########
modules/raft/src/main/java/org/apache/ignite/raft/jraft/error/RaftError.java:
##########
@@ -238,7 +238,12 @@ public enum RaftError {
/**
* Permission denied
*/
- EACCES(1016);
+ EACCES(1016),
+
+ /**
+ * Command reordering detected
+ */
+ EREORDER(1017);
Review Comment:
If we merge JRaft improvements from dora-jraft, it might turn out that they added a new code, and it will conflict with our addition. Is it possible to leave a gap? Like start our specific codes with 2000?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2420,34 +2490,40 @@ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
);
if (!cmd.full()) {
- CompletableFuture<UUID> fut = applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 Temporary code below
+ synchronized (safeTime) {
+ storageUpdateHandler.handleUpdate(
Review Comment:
Why is it that for non-full commands, we apply the command on the Primary first, and only then send the Raft command, but for full commands, the order is opposite?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2371,8 +2410,10 @@ private static <T> boolean allElementsAreNull(List<T> list) {
* @param cmd Raft command.
* @return Raft future.
*/
- private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
- return raftClient.run(cmd).exceptionally(throwable -> {
+ private void applyCmdWithExceptionHandling(Command cmd, CompletableFuture<Object> resultFuture) {
+ applyCmdWithRetryOnSafeTimeReorderException(cmd, resultFuture);
+
+ resultFuture.exceptionally(throwable -> {
Review Comment:
Also this probably means that there are no tests that make sure that timeouts are wrapped in `ReplicationTimeoutException`s (otherwise, they would have failed)
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2371,8 +2410,10 @@ private static <T> boolean allElementsAreNull(List<T> list) {
* @param cmd Raft command.
* @return Raft future.
*/
- private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
- return raftClient.run(cmd).exceptionally(throwable -> {
+ private void applyCmdWithExceptionHandling(Command cmd, CompletableFuture<Object> resultFuture) {
+ applyCmdWithRetryOnSafeTimeReorderException(cmd, resultFuture);
+
+ resultFuture.exceptionally(throwable -> {
Review Comment:
Here, a new stage is added *on top* of `resultFuture`, but this new stage is never returned. It can only 'do something' using its side effects, but it has none. This doesn't look right.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -1677,32 +1703,45 @@ private CompletableFuture<Void> processTxCleanupAction(TxCleanupReplicaRequest r
return reliableCatalogVersionFor(commandTimestamp)
.thenCompose(catalogVersion -> {
- synchronized (commandProcessingLinearizationMutex) {
- TxCleanupCommand txCleanupCmd = MSG_FACTORY.txCleanupCommand()
- .txId(request.txId())
- .commit(request.commit())
- .commitTimestampLong(request.commitTimestampLong())
- .safeTimeLong(hybridClock.nowLong())
- .txCoordinatorId(getTxCoordinatorId(request.txId()))
- .requiredCatalogVersion(catalogVersion)
- .build();
-
- storageUpdateHandler.handleTransactionCleanup(request.txId(), request.commit(), request.commitTimestamp());
-
- raftClient.run(txCleanupCmd)
- .exceptionally(e -> {
- LOG.warn("Failed to complete transaction cleanup command [txId=" + request.txId() + ']', e);
-
- return completedFuture(null);
- });
- }
+ applyCleanupCommand(
+ request.txId(),
+ request.commit(),
+ request.commitTimestamp(),
+ request.commitTimestampLong(),
+ catalogVersion
+ );
return allOffFuturesExceptionIgnored(txReadFutures, request)
.thenRun(() -> releaseTxLocks(request.txId()));
});
});
}
+ private CompletableFuture<Void> applyCleanupCommand(
+ UUID transactionId,
+ boolean commit,
+ HybridTimestamp commitTimestamp,
+ long commitTimestampLong,
+ int catalogVersion
+ ) {
+ TxCleanupCommand txCleanupCmd = MSG_FACTORY.txCleanupCommand()
+ .txId(transactionId)
+ .commit(commit)
+ .commitTimestampLong(commitTimestampLong)
+ .safeTimeLong(hybridClock.nowLong())
+ .txCoordinatorId(getTxCoordinatorId(transactionId))
+ .requiredCatalogVersion(catalogVersion)
+ .build();
+
+ storageUpdateHandler.handleTransactionCleanup(transactionId, commit, commitTimestamp);
+
+ CompletableFuture<Object> resultFuture = new CompletableFuture<>();
+
+ applyCmdWithRetryOnSafeTimeReorderException(txCleanupCmd, resultFuture);
+
+ return resultFuture.thenApply(res -> null);
Review Comment:
Code 'before this change' had `.exceptionally()` to mute an exception happing during cleanup application, now there is no such muting. Is this intentional?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2420,34 +2490,40 @@ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
);
if (!cmd.full()) {
- CompletableFuture<UUID> fut = applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 Temporary code below
+ synchronized (safeTime) {
+ storageUpdateHandler.handleUpdate(
+ cmd.txId(),
+ cmd.rowUuid(),
+ cmd.tablePartitionId().asTablePartitionId(),
+ cmd.rowToUpdate(),
+ true,
+ null,
+ null,
+ null);
+
+ updateTrackerIgnoringTrackerClosedException(safeTime, cmd.safeTime());
+ }
+
+ CompletableFuture<Object> resultFuture = new CompletableFuture<>();
+
+ applyCmdWithExceptionHandling(cmd, resultFuture);
Review Comment:
So we just applied the command on the Primary, and then we send it to Raft with the same SafeTime? Will not this mean that the command will be rejected with probability 100%?
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2383,6 +2424,35 @@ private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
});
}
+ private void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<Object> resultFuture) {
+ raftClient.run(cmd).whenComplete((res, ex) -> {
+ if (ex != null) {
+ if (ex instanceof SafeTimeReorderException || ex.getCause() instanceof SafeTimeReorderException) {
+ assert cmd instanceof SafeTimePropagatingCommand;
+
+ SafeTimePropagatingCommand safeTimePropagatingCommand = (SafeTimePropagatingCommand) cmd;
+
+ HybridTimestamp safeTimeForRetry = hybridClock.now();
+
+ if ((cmd instanceof UpdateCommand && !((UpdateCommand) cmd).full())
Review Comment:
Also, this check seems to be fragile. What if we add an analogous request in the future, but forget to mention it here?
How about adding a marker interface common for both update commands? It might be more difficult to forget about it when adding a third 'update' command in the future.
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2383,6 +2424,35 @@ private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
});
}
+ private void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<Object> resultFuture) {
+ raftClient.run(cmd).whenComplete((res, ex) -> {
+ if (ex != null) {
+ if (ex instanceof SafeTimeReorderException || ex.getCause() instanceof SafeTimeReorderException) {
+ assert cmd instanceof SafeTimePropagatingCommand;
Review Comment:
This line probably means that the type of the parameter can be safely changed to `SafeTimePropagatingCommand`, making both the assertion and the following cast obsolete
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -455,6 +478,21 @@ public void onShutdown() {
storage.close();
}
+ @Override
+ public void onBeforeApply(Command command) {
+ // Given method is is synchronized by replication group specific monitor, see ActionRequestProcessor#handleRequest.
Review Comment:
```suggestion
// This method is synchronized by replication group specific monitor, see ActionRequestProcessor#handleRequest.
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2383,6 +2424,35 @@ private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
});
}
+ private void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<Object> resultFuture) {
+ raftClient.run(cmd).whenComplete((res, ex) -> {
+ if (ex != null) {
+ if (ex instanceof SafeTimeReorderException || ex.getCause() instanceof SafeTimeReorderException) {
+ assert cmd instanceof SafeTimePropagatingCommand;
+
+ SafeTimePropagatingCommand safeTimePropagatingCommand = (SafeTimePropagatingCommand) cmd;
+
+ HybridTimestamp safeTimeForRetry = hybridClock.now();
+
+ if ((cmd instanceof UpdateCommand && !((UpdateCommand) cmd).full())
Review Comment:
Why is this condition needed? Could you please add a comment explaining this?
--
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-20709 [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389054077
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2420,34 +2490,40 @@ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
);
if (!cmd.full()) {
- CompletableFuture<UUID> fut = applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 Temporary code below
+ synchronized (safeTime) {
+ storageUpdateHandler.handleUpdate(
+ cmd.txId(),
+ cmd.rowUuid(),
+ cmd.tablePartitionId().asTablePartitionId(),
+ cmd.rowToUpdate(),
+ true,
+ null,
+ null,
+ null);
+
+ updateTrackerIgnoringTrackerClosedException(safeTime, cmd.safeTime());
+ }
+
+ CompletableFuture<Object> resultFuture = new CompletableFuture<>();
+
+ applyCmdWithExceptionHandling(cmd, resultFuture);
Review Comment:
So we just applied the command on the Primary, and then we send it to Raft with the same SafeTime? Will not this mean that the command will be rejected with probability equal to 100%?
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc merged PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1390932001
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2420,34 +2490,40 @@ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
);
if (!cmd.full()) {
- CompletableFuture<UUID> fut = applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 Temporary code below
+ synchronized (safeTime) {
+ storageUpdateHandler.handleUpdate(
Review Comment:
Replication is async anyway (it returns a future which we don't wait). For !1PC, before your fix it was (schematically):
```
var future = replicateCommand(cmd);
applyLocally(cmd);
return future;
```
Now it is
```
applyLocally(cmd);
var future = replicateCommand(cmd);
return future;
```
It seems that we did not wait for replication in the previous version of the code as well. But the order has changed. Why? :)
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1390775566
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2420,34 +2490,40 @@ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
);
if (!cmd.full()) {
- CompletableFuture<UUID> fut = applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 Temporary code below
+ synchronized (safeTime) {
+ storageUpdateHandler.handleUpdate(
Review Comment:
For better performance. We will return each operation to the user right after handling within primary replica and we will wait for all replication actions to be completed asynchronously before commit.
--
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-20709 [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389599335
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2383,6 +2424,35 @@ private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
});
}
+ private void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<Object> resultFuture) {
+ raftClient.run(cmd).whenComplete((res, ex) -> {
+ if (ex != null) {
+ if (ex instanceof SafeTimeReorderException || ex.getCause() instanceof SafeTimeReorderException) {
+ assert cmd instanceof SafeTimePropagatingCommand;
+
+ SafeTimePropagatingCommand safeTimePropagatingCommand = (SafeTimePropagatingCommand) cmd;
+
+ HybridTimestamp safeTimeForRetry = hybridClock.now();
+
+ if ((cmd instanceof UpdateCommand && !((UpdateCommand) cmd).full())
+ || (cmd instanceof UpdateAllCommand && !((UpdateAllCommand) cmd).full())) {
+ synchronized (safeTime) {
+ updateTrackerIgnoringTrackerClosedException(safeTime, safeTimeForRetry);
+ }
+ }
+
+ safeTimePropagatingCommand.safeTimeLong(safeTimeForRetry.longValue());
+
+ applyCmdWithRetryOnSafeTimeReorderException(safeTimePropagatingCommand, resultFuture);
Review Comment:
Attempts counter along with limit added.
--
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-20709 [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389054077
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2420,34 +2490,40 @@ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
);
if (!cmd.full()) {
- CompletableFuture<UUID> fut = applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 Temporary code below
+ synchronized (safeTime) {
+ storageUpdateHandler.handleUpdate(
+ cmd.txId(),
+ cmd.rowUuid(),
+ cmd.tablePartitionId().asTablePartitionId(),
+ cmd.rowToUpdate(),
+ true,
+ null,
+ null,
+ null);
+
+ updateTrackerIgnoringTrackerClosedException(safeTime, cmd.safeTime());
+ }
+
+ CompletableFuture<Object> resultFuture = new CompletableFuture<>();
+
+ applyCmdWithExceptionHandling(cmd, resultFuture);
Review Comment:
~So we just applied the command on the Primary, and then we send it to Raft with the same SafeTime? Will not this mean that the command will be rejected with probability equal to 100%?~
No, that's not the case as `PartitionListener` looks at its local 'last observed SafeTime ts'.
--
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-20709 [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389466876
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2371,8 +2410,10 @@ private static <T> boolean allElementsAreNull(List<T> list) {
* @param cmd Raft command.
* @return Raft future.
*/
- private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
- return raftClient.run(cmd).exceptionally(throwable -> {
+ private void applyCmdWithExceptionHandling(Command cmd, CompletableFuture<Object> resultFuture) {
+ applyCmdWithRetryOnSafeTimeReorderException(cmd, resultFuture);
+
+ resultFuture.exceptionally(throwable -> {
Review Comment:
Good catch! Fixed.
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "rpuch (via GitHub)" <gi...@apache.org>.
rpuch commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1390716173
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -938,7 +942,8 @@ private CompletableFuture<Void> processReplicaSafeTimeSyncRequest(ReplicaSafeTim
applyCmdWithRetryOnSafeTimeReorderException(
REPLICA_MESSAGES_FACTORY.safeTimeSyncCommand().safeTimeLong(hybridClock.nowLong()).build(),
- resultFuture
+ resultFuture,
+ 0
Review Comment:
How about adding an overload with 2 arguments that just calls the 3-arg version passing 0 as the last arg? This would probably make the initial invocations a bit cleaner.
--
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-20709 [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389421124
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -99,6 +101,15 @@ public class PartitionListener implements RaftGroupListener {
/** Storage index tracker. */
private final PendingComparableValuesTracker<Long, Void> storageIndexTracker;
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20826 Restore on restart
+ /** Is used in order to detect and retry safe time reordering within onBeforeApply. */
+ private long maxObservableSafeTime = -1;
+
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20826 Restore on restart
+ /** Is used in order to assert safe time reordering within onWrite. */
+ private long maxObservableSafeTimeVerificator = -1;
Review Comment:
Fixed.
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1390790340
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2420,34 +2490,40 @@ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
);
if (!cmd.full()) {
- CompletableFuture<UUID> fut = applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 Temporary code below
+ synchronized (safeTime) {
+ storageUpdateHandler.handleUpdate(
+ cmd.txId(),
+ cmd.rowUuid(),
+ cmd.tablePartitionId().asTablePartitionId(),
+ cmd.rowToUpdate(),
+ true,
+ null,
+ null,
+ null);
+
+ updateTrackerIgnoringTrackerClosedException(safeTime, cmd.safeTime());
+ }
+
+ CompletableFuture<Object> resultFuture = new CompletableFuture<>();
+
+ applyCmdWithExceptionHandling(cmd, resultFuture);
Review Comment:
It's just a statement and you do not expect any comments here, right?
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1390939407
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2420,34 +2490,40 @@ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
);
if (!cmd.full()) {
- CompletableFuture<UUID> fut = applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 Temporary code below
+ synchronized (safeTime) {
+ storageUpdateHandler.handleUpdate(
Review Comment:
Because in new version it's guaranteed that we will update storage and safeTime prior to replication. Before there was a race 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-20709 [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389421582
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/raft/PartitionListener.java:
##########
@@ -455,6 +478,21 @@ public void onShutdown() {
storage.close();
}
+ @Override
+ public void onBeforeApply(Command command) {
+ // Given method is is synchronized by replication group specific monitor, see ActionRequestProcessor#handleRequest.
Review Comment:
Fixed.
--
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-20709 [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389504176
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2383,6 +2424,35 @@ private CompletableFuture<Object> applyCmdWithExceptionHandling(Command cmd) {
});
}
+ private void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, CompletableFuture<Object> resultFuture) {
+ raftClient.run(cmd).whenComplete((res, ex) -> {
+ if (ex != null) {
+ if (ex instanceof SafeTimeReorderException || ex.getCause() instanceof SafeTimeReorderException) {
+ assert cmd instanceof SafeTimePropagatingCommand;
+
+ SafeTimePropagatingCommand safeTimePropagatingCommand = (SafeTimePropagatingCommand) cmd;
+
+ HybridTimestamp safeTimeForRetry = hybridClock.now();
+
+ if ((cmd instanceof UpdateCommand && !((UpdateCommand) cmd).full())
Review Comment:
> Why is this condition needed? Could you please add a comment explaining this?
Added.
```
// Within primary replica it's required to update safe time in order to prevent double storage updates in case of !1PC.
// Otherwise, it may be possible that a newer entry will be overwritten by an older one that came as part of the raft
// replication flow:
// tx1 = transactions.begin();
// tx1.put(k1, v1) -> primary.apply(k1,v1) + asynchronous raft replication (k1,v1)
// tx1.put(k1, v2) -> primary.apply(k1,v2) + asynchronous raft replication (k1,v1)
// (k1,v1) replication overrides newer (k1, v2). Eventually (k1,v2) replication will restore proper value.
// However it's possible that tx1.get(k1) will see v1 instead of v2.
```
>
> Also, this check seems to be fragile. What if we add an analogous request in the future, but forget to mention it here?
>
> How about adding a marker interface common for both update commands? It might be more difficult to forget about it when adding a third 'update' command in the future.
I agree that given solution is fragile. It should be refactored with something more reliable in https://issues.apache.org/jira/browse/IGNITE-20124 TODO added. Here I'm talking not only about safeTime adjustment on retry but about general double updates prevention logic.
--
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-20709 [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1389522573
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2420,34 +2490,40 @@ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
);
if (!cmd.full()) {
- CompletableFuture<UUID> fut = applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 Temporary code below
+ synchronized (safeTime) {
+ storageUpdateHandler.handleUpdate(
Review Comment:
Because in case of full command we apply not the write intent but write itself. Writes doesn't require txn state resolution and thus visible to other transactions directly. All in all, that means that we should prevent write visibility in case of replication failure. The reliable way to do it is to put replication prior to primary replica adjustment.
Why we need direct put on primary replica?
Not only because replication is finished when it's finished on majoirty that may miss primary but also because primary replica may not be a part of replication group.
--
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-20709 Fix safe time reordering wihtin partitions [ignite-3]
Posted by "sanpwc (via GitHub)" <gi...@apache.org>.
sanpwc commented on code in PR #2820:
URL: https://github.com/apache/ignite-3/pull/2820#discussion_r1390804600
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -938,7 +942,8 @@ private CompletableFuture<Void> processReplicaSafeTimeSyncRequest(ReplicaSafeTim
applyCmdWithRetryOnSafeTimeReorderException(
REPLICA_MESSAGES_FACTORY.safeTimeSyncCommand().safeTimeLong(hybridClock.nowLong()).build(),
- resultFuture
+ resultFuture,
+ 0
Review Comment:
Agree. Fixed.
--
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