rpuch commented on code in PR #5241: URL: https://github.com/apache/ignite-3/pull/5241#discussion_r1963420986
########## 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: We don't need `clock.now()` here, I've changed the way we calculate the timestamp (and found a potential problem, I've filed https://issues.apache.org/jira/browse/IGNITE-24574 and added TODOs). Also, I made the scope under `reliableCatalogVersion()` as small as possible. But non of these fixes actually address your main concerns outlined in this comment, so I'm still working on it. >It's fragile that PartitionReplicaListener.processWriteIntentSwitchAction is just a method and not a Handler How is it fragile? >WriteIntentTableSwitchCommand Would it be used for per-zone partitions (so generated in `ZonePartitionReplicaListener`)? -- 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
