diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteStripedReadWriteLock.java b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteStripedReadWriteLock.java index 2936149cc44..b799cc59f8b 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteStripedReadWriteLock.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteStripedReadWriteLock.java @@ -31,7 +31,7 @@ */ public class IgniteStripedReadWriteLock implements ReadWriteLock { /** Default concurrency. */ - public static int CONCURRENCY = Math.max(1, Runtime.getRuntime().availableProcessors() / 2); + private static final int CONCURRENCY = Math.max(1, Runtime.getRuntime().availableProcessors() / 2); /** Index generator. */ private static final AtomicInteger IDX_GEN = new AtomicInteger(); diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java index 15505744ea8..392f338f005 100644 --- a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java +++ b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java @@ -1738,7 +1738,7 @@ private CompletableFuture finishTransaction( throw new TransactionException(commit ? TX_COMMIT_ERR : TX_ROLLBACK_ERR, ex); } - TransactionResult result = (TransactionResult) ((ApplyCommandResult) txOutcome).result; + TransactionResult result = (TransactionResult) ((ResultWrapper) txOutcome).result; markFinished(txId, result.transactionState(), result.commitTimestamp()); @@ -2620,8 +2620,8 @@ private static boolean allElementsAreNull(List list) { * @param cmd Raft command. * @return Raft future or raft decorated future with command that was processed. */ - private CompletableFuture> applyCmdWithExceptionHandling(Command cmd) { - CompletableFuture> resultFuture = new CompletableFuture<>(); + private CompletableFuture> applyCmdWithExceptionHandling(Command cmd) { + CompletableFuture> resultFuture = new CompletableFuture<>(); applyCmdWithRetryOnSafeTimeReorderException(cmd, resultFuture); @@ -2677,7 +2677,7 @@ private void applyCmdWithRetryOnSafeTimeReorderException(Command cmd, Comple resultFuture.completeExceptionally(ex); } } else { - resultFuture.complete((T) new ApplyCommandResult<>(cmd, res)); + resultFuture.complete((T) new ResultWrapper<>(cmd, res)); } }); } @@ -2751,7 +2751,7 @@ private CompletableFuture applyUpdateCommand( if (updateCommandResult != null && updateCommandResult.isPrimaryInPeersAndLearners()) { return safeTime.waitFor(((UpdateCommand) res.getCommand()).safeTime()).thenApply(ignored -> null) - .thenApply(ret -> new ApplyResult(cmd.safeTime(), null)); + .thenApply(ret -> new ApplyResult(((UpdateCommand) res.getCommand()).safeTime(), null)); } else { if (!SKIP_UPDATES) { // We don't need to take the partition snapshots read lock, see #INTERNAL_DOC_PLACEHOLDER why. @@ -2768,7 +2768,8 @@ private CompletableFuture applyUpdateCommand( ); } - return completedFuture(new ApplyResult(cmd.safeTime(), null)); + // getCommand provides actual assigned safeTime (may be reassigned due to reorder) + return completedFuture(new ApplyResult(((UpdateCommand) res.getCommand()).safeTime(), null)); } }); } @@ -2885,7 +2886,7 @@ private CompletableFuture applyUpdateAllCommand( } if (updateCommandResult.isPrimaryInPeersAndLearners()) { return safeTime.waitFor(((UpdateAllCommand) res.getCommand()).safeTime()) - .thenApply(ret -> new ApplyResult(cmd.safeTime(), null)); + .thenApply(ret -> new ApplyResult(((UpdateAllCommand) res.getCommand()).safeTime(), null)); } else { // We don't need to take the partition snapshots read lock, see #INTERNAL_DOC_PLACEHOLDER why. storageUpdateHandler.handleUpdateAll( @@ -2898,7 +2899,7 @@ private CompletableFuture applyUpdateAllCommand( indexIdsAtRwTxBeginTs(txId) ); - return completedFuture(new ApplyResult(cmd.safeTime(), null)); + return completedFuture(new ApplyResult(((UpdateAllCommand) res.getCommand()).safeTime(), null)); } }); } @@ -4147,11 +4148,11 @@ private static Map asTablePartitionIdStringMap(Map { + private static class ResultWrapper { private final Command command; private final T result; - ApplyCommandResult(Command command, T result) { + ResultWrapper(Command command, T result) { this.command = command; this.result = result; }