sanpwc commented on code in PR #5241: URL: https://github.com/apache/ignite-3/pull/5241#discussion_r1962067729
########## modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/WriteIntentSwitchRequestHandler.java: ########## @@ -0,0 +1,135 @@ +/* + * 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.partition.replicator; + +import static java.util.concurrent.CompletableFuture.allOf; +import static org.apache.ignite.internal.tx.TxState.ABORTED; +import static org.apache.ignite.internal.tx.TxState.COMMITTED; + +import java.util.UUID; +import java.util.concurrent.CompletableFuture; +import java.util.function.IntFunction; +import org.apache.ignite.internal.catalog.CatalogService; +import org.apache.ignite.internal.hlc.ClockService; +import org.apache.ignite.internal.logger.IgniteLogger; +import org.apache.ignite.internal.logger.Loggers; +import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessagesFactory; +import org.apache.ignite.internal.partition.replicator.network.command.WriteIntentSwitchCommand; +import org.apache.ignite.internal.raft.service.RaftCommandRunner; +import org.apache.ignite.internal.replicator.CommandApplicationResult; +import org.apache.ignite.internal.replicator.ReplicaResult; +import org.apache.ignite.internal.replicator.ZonePartitionId; +import org.apache.ignite.internal.replicator.listener.ReplicaListener; +import org.apache.ignite.internal.schema.SchemaSyncService; +import org.apache.ignite.internal.tx.TxManager; +import org.apache.ignite.internal.tx.TxState; +import org.apache.ignite.internal.tx.message.WriteIntentSwitchReplicaRequest; +import org.apache.ignite.internal.tx.message.WriteIntentSwitchReplicatedInfo; + +/** + * Handles {@link WriteIntentSwitchReplicaRequest}s. + */ +public class WriteIntentSwitchRequestHandler { + private static final IgniteLogger LOG = Loggers.forClass(WriteIntentSwitchRequestHandler.class); + + private static final PartitionReplicationMessagesFactory PARTITION_REPLICATION_MESSAGES_FACTORY = + new PartitionReplicationMessagesFactory(); + + private final IntFunction<ReplicaListener> replicaListenerByTableId; + + private final ClockService clockService; + + private final ZonePartitionId replicationGroupId; + + private final ReliableCatalogVersions reliableCatalogVersions; + private final ReplicaTxFinishMarker txFinishMarker; + private final ReplicationRaftCommandApplicator raftCommandApplicator; + + /** Constructor. */ + public WriteIntentSwitchRequestHandler( + IntFunction<ReplicaListener> replicaListenerByTableId, + ClockService clockService, + SchemaSyncService schemaSyncService, + CatalogService catalogService, + TxManager txManager, + RaftCommandRunner raftCommandRunner, + ZonePartitionId replicationGroupId + ) { + this.replicaListenerByTableId = replicaListenerByTableId; + this.clockService = clockService; + this.replicationGroupId = replicationGroupId; + + reliableCatalogVersions = new ReliableCatalogVersions(schemaSyncService, catalogService); + txFinishMarker = new ReplicaTxFinishMarker(txManager); + raftCommandApplicator = new ReplicationRaftCommandApplicator(raftCommandRunner, replicationGroupId); + } + + /** + * Handles write intent switch requests: + * <ol> + * <li>Waits for finishing of local transactional operations;</li> + * <li>Runs asynchronously the specific raft {@code WriteIntentSwitchCommand} command, that will convert all pending entries + * (writeIntents) to either regular values({@link TxState#COMMITTED}) or removing them ({@link TxState#ABORTED});</li> + * <li>Releases all locks that were held on local Replica by given transaction.</li> + * </ol> + * This operation is idempotent, so it's safe to retry it. + * + * @param request Transaction cleanup request. + * @return CompletableFuture of ReplicaResult. + */ + public CompletableFuture<ReplicaResult> handle(WriteIntentSwitchReplicaRequest request, UUID senderId) { + txFinishMarker.markFinished(request.txId(), request.commit() ? COMMITTED : ABORTED, request.commitTimestamp()); + + @SuppressWarnings("rawtypes") + CompletableFuture[] futures = request.tableIds().stream() + .map(tableId -> replicaListener(tableId).invoke(request, senderId)) + .toArray(CompletableFuture[]::new); + + return allOf(futures) + .thenCompose(unused -> reliableCatalogVersions.reliableCatalogVersionFor(clockService.now())) Review Comment: - Every tableProcessor will evaluates commandTimestamp as `clockService.now()` along with ZonePartitionReplicaListener. That's excessive - I do remember that Alexey spend some time removing unessary clockService.now() calls. - Besides that we duplicate reliableCatalogVersionFor calls. - Inside PartitionReplicaListener we will process WISR only if `awaitCleanupReadyFutures` will return result that `(res.hadUpdateFutures() || res.forceCleanup())`. There's no such logic in WriteIntentSwitchRequestHandler. Why? - It's fragile that PartitionReplicaListener.processWriteIntentSwitchAction is just a method and not a Handler. I'd further differentiate this logic by introducing special command WriteIntentTableSwitchCommand in order to propagate reliableCatalogVersions etc and put processing into handler. -- 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-unsubscr...@ignite.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org
