sanpwc commented on code in PR #2689:
URL: https://github.com/apache/ignite-3/pull/2689#discussion_r1360846918
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -2038,133 +2094,195 @@ private CompletableFuture<Object>
applyCmdWithExceptionHandling(Command cmd) {
/**
* Executes an Update command.
*
- * @param cmd Update command.
+ * @param tablePartId {@link TablePartitionId} object.
+ * @param rowUuid Row UUID.
+ * @param row Row.
+ * @param lastCommitTimestamp The timestamp of the last committed entry
for the row.
+ * @param txId Transaction ID.
+ * @param full {@code True} if this is a full transaction.
+ * @param txCoordinatorId Transaction coordinator id.
+ * @param catalogVersion Validated catalog version associated with given
operation.
* @return A local update ready future, possibly having a nested
replication future as a result for delayed ack purpose.
*/
- private CompletableFuture<CompletableFuture<?>>
applyUpdateCommand(UpdateCommand cmd) {
- if (!cmd.full()) {
- CompletableFuture<UUID> fut =
applyCmdWithExceptionHandling(cmd).thenApply(res -> {
- // This check guaranties the result will never be lost.
Currently always null.
- assert res == null : "Replication result is lost";
-
- // Set context for delayed response.
- return cmd.txId();
- });
-
- storageUpdateHandler.handleUpdate(
- cmd.txId(),
- cmd.rowUuid(),
- cmd.tablePartitionId().asTablePartitionId(),
- cmd.row(),
- true,
- null,
- null,
- cmd.lastCommitTimestamp());
-
- // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 tmp
- synchronized (safeTime) {
- updateTrackerIgnoringTrackerClosedException(safeTime,
cmd.safeTime());
- }
+ private CompletableFuture<CompletableFuture<?>> applyUpdateCommand(
+ TablePartitionId tablePartId,
+ UUID rowUuid,
+ @Nullable BinaryRow row,
+ @Nullable HybridTimestamp lastCommitTimestamp,
+ UUID txId,
+ boolean full,
+ String txCoordinatorId,
+ int catalogVersion
+ ) {
+ synchronized (commandProcessingLinearizationMutex) {
+ UpdateCommand cmd = updateCommand(
+ tablePartId,
+ rowUuid,
+ row,
+ lastCommitTimestamp,
+ txId,
+ full,
+ txCoordinatorId,
+ hybridClock.now(),
+ catalogVersion
+ );
+
+ if (!cmd.full()) {
+ CompletableFuture<UUID> fut =
applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // This check guaranties the result will never be lost.
Currently always null.
+ assert res == null : "Replication result is lost";
- return completedFuture(fut);
- } else {
- return applyCmdWithExceptionHandling(cmd).thenApply(res -> {
- // This check guaranties the result will never be lost.
Currently always null.
- assert res == null : "Replication result is lost";
+ // Set context for delayed response.
+ return cmd.txId();
+ });
- // Try to avoid double write if an entry is already replicated.
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-20124 tmp
synchronized (safeTime) {
if (cmd.safeTime().compareTo(safeTime.current()) > 0) {
storageUpdateHandler.handleUpdate(
cmd.txId(),
cmd.rowUuid(),
cmd.tablePartitionId().asTablePartitionId(),
cmd.row(),
- false,
+ true,
+ null,
null,
- cmd.safeTime(),
cmd.lastCommitTimestamp());
updateTrackerIgnoringTrackerClosedException(safeTime,
cmd.safeTime());
}
}
- return null;
- });
+ return completedFuture(fut);
+ } else {
+ return applyCmdWithExceptionHandling(cmd).thenApply(res -> {
+ // This check guaranties the result will never be lost.
Currently always null.
+ assert res == null : "Replication result is lost";
+
+ // TODO:
https://issues.apache.org/jira/browse/IGNITE-20124 tmp
Review Comment:
tmp here means that code below is temporary and will be adjusted within
IGNITE-20124. Will rewrite the comment.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
