You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "vldpyatkov (via GitHub)" <gi...@apache.org> on 2023/11/10 20:07:31 UTC
[PR] Ignite 20685 2 [ignite-3]
vldpyatkov opened a new pull request, #2832:
URL: https://github.com/apache/ignite-3/pull/2832
https://issues.apache.org/jira/browse/IGNITE-20685
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1395958562
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.tx.impl;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The class detects transactions that are left without a coordinator but still hold locks. For that orphan transaction, the recovery
+ * message is sent to the commit partition replication group.
+ */
+public class OrphanDetector {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(OrphanDetector.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ private static final long AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC = 10;
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ private final PlacementDriver placementDriver;
+
+ private final HybridClock clock;
+
+ /** The local map for tx states. */
+ private ConcurrentHashMap<UUID, TxStateMeta> txStateMap;
+
+ /**
+ * The constructor.
+ *
+ * @param clusterService Cluster service.
+ * @param placementDriver Placement driver.
+ * @param clock Clock.
+ */
+ public OrphanDetector(ClusterService clusterService, PlacementDriver placementDriver, HybridClock clock) {
+ this.clusterService = clusterService;
+ this.placementDriver = placementDriver;
+ this.clock = clock;
+ }
+
+
+ /**
+ * Starts the detector.
+ *
+ * @param txStateMap Transaction state map.
+ */
+ public void start(ConcurrentHashMap<UUID, TxStateMeta> txStateMap) {
+ this.txStateMap = txStateMap;
+ // Subscribe to lock conflicts here.
+ }
Review Comment:
Ok
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1392532552
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMeta.java:
##########
@@ -30,22 +33,28 @@ public class TxStateMeta implements TransactionMeta {
private final String txCoordinatorId;
+ /** Identifier of the replication group that manages a transaction state. */
+ private ReplicationGroupId commitPartitionId;
+
private final HybridTimestamp commitTimestamp;
/**
* Constructor.
*
* @param txState Transaction state.
* @param txCoordinatorId Transaction coordinator id.
+ * @param commitPartitionId Commit partition replication group id.
* @param commitTimestamp Commit timestamp.
*/
public TxStateMeta(
TxState txState,
String txCoordinatorId,
+ ReplicationGroupId commitPartitionId,
Review Comment:
Please mark it as `@Nullable`
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1393828241
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.tx.impl;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The class detects transactions that are left without a coordinator but still hold locks. For that orphan transaction, the recovery
+ * message is sent to the commit partition replication group.
+ */
+public class OrphanDetector {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(OrphanDetector.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ private static final long AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC = 10;
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ private final PlacementDriver placementDriver;
+
+ private final HybridClock clock;
+
+ /** The local map for tx states. */
+ private ConcurrentHashMap<UUID, TxStateMeta> txStateMap;
+
+ /**
+ * The constructor.
+ *
+ * @param clusterService Cluster service.
+ * @param placementDriver Placement driver.
+ * @param clock Clock.
+ */
+ public OrphanDetector(ClusterService clusterService, PlacementDriver placementDriver, HybridClock clock) {
+ this.clusterService = clusterService;
+ this.placementDriver = placementDriver;
+ this.clock = clock;
+ }
+
+
+ /**
+ * Starts the detector.
+ *
+ * @param txStateMap Transaction state map.
+ */
+ public void start(ConcurrentHashMap<UUID, TxStateMeta> txStateMap) {
+ this.txStateMap = txStateMap;
+ // Subscribe to lock conflicts here.
+ }
+
+ /**
+ * Stops the detector.
+ */
+ public void stop() {
+ busyLock.block();
+ // Unsubscribe from lock conflicts here.
+ }
+
+ /**
+ * Sends {@link TxRecoveryMessage} if the transaction is orphaned.
+ *
+ * @param txId Transaction id that holds a lock.
+ * @return Future to complete.
+ */
+ private CompletableFuture<Void> handleLockHolder(UUID txId) {
Review Comment:
Yes, it would be called by the event from Lock Manager.
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1393826466
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.tx.impl;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The class detects transactions that are left without a coordinator but still hold locks. For that orphan transaction, the recovery
+ * message is sent to the commit partition replication group.
+ */
+public class OrphanDetector {
Review Comment:
I think it is a detector because it checks that the transation coordinator is absent in topology.
```
clusterService.topologyService().getById(txState.txCoordinatorId()) == null
```
It would be triggered by Lock Manager, but it is a task for another ticket.
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1392544739
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/HeapLockManager.java:
##########
@@ -167,7 +167,7 @@ public Waiter waiter(LockKey key, UUID txId) {
/**
* A lock state.
*/
- private static class LockState {
Review Comment:
why is it not static anymore?
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1398885722
##########
modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java:
##########
@@ -433,14 +437,26 @@ public TxManager txManager() {
* @param replicaSvc Replica service to use.
*/
public static TxManagerImpl txManager(ReplicaService replicaSvc) {
- return new TxManagerImpl(
+ TopologyService topologyService = mock(TopologyService.class);
+ when(topologyService.localMember()).thenReturn(LOCAL_NODE);
+
+ ClusterService clusterService = mock(ClusterService.class);
+ when(clusterService.messagingService()).thenReturn(mock(MessagingService.class));
+ when(clusterService.topologyService()).thenReturn(topologyService);
+
+ var txManager = new TxManagerImpl(
+ clusterService,
replicaSvc,
new HeapLockManager(),
CLOCK,
new TransactionIdGenerator(0xdeadbeef),
- LOCAL_NODE::id,
- TEST_PLACEMENT_DRIVER
+ TEST_PLACEMENT_DRIVER,
+ () -> DEFAULT_IDLE_SAFE_TIME_PROPAGATION_PERIOD_MILLISECONDS
);
+
+ txManager.start();
Review Comment:
You are right, but it is just a test. Lackly, nothing was violated when the manager started without stopping.
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1392565761
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryProcessor.java:
##########
@@ -0,0 +1,109 @@
+/*
+ * 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.tx.impl;
+
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The processor receives transaction recovery messages. The recovery message is received when the node has a commit partition for the
+ * specific transaction.
+ */
+public class TxRecoveryProcessor {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryProcessor.class);
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The local map for tx states. */
+ private ConcurrentHashMap<UUID, TxStateMeta> txStateMap;
+
+ /**
+ * The constructor.
+ *
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryProcessor(ClusterService clusterService) {
+ this.clusterService = clusterService;
+ }
+
+ /**
+ * Starts the recovery processor.
+ *
+ * @param txStateMap Transaction state map.
+ */
+ public void start(ConcurrentHashMap<UUID, TxStateMeta> txStateMap) {
+ this.txStateMap = txStateMap;
+
+ clusterService.messagingService().addMessageHandler(TxMessageGroup.class, (msg, sender, correlationId) -> {
+ if (msg instanceof TxRecoveryMessage) {
+ if (busyLock.enterBusy()) {
+ try {
+ processTxRecoveryMessage((TxRecoveryMessage) msg);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ } else {
+ LOG.info("Transaction recovery message was ignored [msg={}]", msg);
+ }
+ }
+ });
+ }
+
+ /**
+ * Processes the recovery transaction message.
+ *
+ * @param msg Transaction recovery message.
+ */
+ private void processTxRecoveryMessage(TxRecoveryMessage msg) {
Review Comment:
Okay, I checked the code again and see that this message is sent onto the commit partition primary only. This approach looks different from the other messages handling (which is done in PartitionReplicaListener). Is there any specific reason we have this class?
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1394106184
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxRecoveryMessage.java:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.tx.message;
+
+import java.util.UUID;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.annotations.Transferable;
+
+/**
+ * Transaction recovery message.
+ */
+@Transferable(TxMessageGroup.TX_RECOVERY_MSG)
+public interface TxRecoveryMessage extends NetworkMessage {
+ /**
+ * Gets a transaction id to resolve.
+ *
+ * @return Transaction id.
+ */
Review Comment:
As per the initial discussion, the commit partition primary after processing the recovery message sends a cleanup message back to the request initiator.
If we still expect this functionality then we need initiator's TablePartitionId
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1395959946
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMetaFinishing.java:
##########
@@ -35,9 +36,10 @@ public class TxStateMetaFinishing extends TxStateMeta {
* Constructor.
*
* @param txCoordinatorId Transaction coordinator id.
+ * @param commitPartitionId Commit partition id.
*/
- public TxStateMetaFinishing(String txCoordinatorId) {
- super(TxState.FINISHING, txCoordinatorId, null);
+ public TxStateMetaFinishing(String txCoordinatorId, ReplicationGroupId commitPartitionId) {
Review Comment:
Done.
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1392421835
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/HeapLockManager.java:
##########
@@ -179,13 +179,16 @@ private static class LockState {
/** Marked for removal flag. */
private boolean markedForRemove = false;
- public LockState(DeadlockPreventionPolicy deadlockPreventionPolicy, Executor delayedExecutor) {
+ private final LockKey lockKey;
Review Comment:
seems that this field in not used
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItTransactionConflictTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.table;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
+import static org.apache.ignite.internal.util.ExceptionUtils.extractCodeFrom;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.lang.ErrorGroups.Transactions;
+import org.apache.ignite.table.RecordView;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.tx.Transaction;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+
+/**
+ * Abundant transactions integration tests.
Review Comment:
Did you mean "abandoned"?
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java:
##########
@@ -349,9 +332,17 @@ public CompletableFuture<Void> finish(
// than all the read timestamps processed before.
// Every concurrent operation will now use a finish future from the finishing state meta and get only final transaction
// state after the transaction is finished.
- TxStateMetaFinishing finishingStateMeta = new TxStateMetaFinishing(coordinatorId());
+ AtomicReference<TxStateMetaFinishing> finishingStateMetaRef = new AtomicReference<>();
+
+ updateTxMeta(txId, old -> {
+ var finishingState = new TxStateMetaFinishing(localNodeId, old == null ? null : old.commitPartitionId());
- updateTxMeta(txId, old -> finishingStateMeta);
+ finishingStateMetaRef.set(finishingState);
+
+ return finishingState;
+ });
Review Comment:
I think it would be better to make `updateTxMeta` return the state that was actually set.
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMeta.java:
##########
@@ -30,22 +33,28 @@ public class TxStateMeta implements TransactionMeta {
private final String txCoordinatorId;
+ /** Identifier of the replication group that manages a transaction state. */
+ private ReplicationGroupId commitPartitionId;
Review Comment:
`TablePartitionId` would be more useful here.
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyTransactionImpl.java:
##########
@@ -116,7 +116,7 @@ public CompletableFuture<Void> finish(boolean commit, HybridTimestamp executionT
return ((TxManagerImpl) txManager).completeReadOnlyTransactionFuture(new TxIdAndTimestamp(readTimestamp, id()))
.thenRun(() -> txManager.updateTxMeta(
id(),
- old -> new TxStateMeta(COMMITED, old.txCoordinatorId(), old.commitTimestamp())
+ old -> new TxStateMeta(COMMITED, old.txCoordinatorId(), null, old.commitTimestamp())
Review Comment:
The commit partition should be taken from the old value here, otherwise it can be lost
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItTransactionConflictTest.java:
##########
@@ -0,0 +1,162 @@
+/*
+ * 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.table;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.SessionUtils.executeUpdate;
+import static org.apache.ignite.internal.testframework.IgniteTestUtils.waitForCondition;
+import static org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willCompleteSuccessfully;
+import static org.apache.ignite.internal.util.ExceptionUtils.extractCodeFrom;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.stream.IntStream;
+import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.placementdriver.ReplicaMeta;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.lang.ErrorGroups.Transactions;
+import org.apache.ignite.table.RecordView;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.tx.Transaction;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInfo;
+
+/**
+ * Abundant transactions integration tests.
+ */
+public class ItTransactionConflictTest extends ClusterPerTestIntegrationTest {
+ /** Table name. */
+ private static final String TABLE_NAME = "test_table";
+
+ @BeforeEach
+ @Override
+ public void setup(TestInfo testInfo) throws Exception {
+ super.setup(testInfo);
+
+ String zoneSql = "create zone test_zone with partitions=1, replicas=3";
+ String sql = "create table " + TABLE_NAME + " (key int primary key, val varchar(20)) with primary_zone='TEST_ZONE'";
+
+ cluster.doInSession(0, session -> {
+ executeUpdate(zoneSql, session);
+ executeUpdate(sql, session);
+ });
+ }
+
+ @Test
+ @Disabled("https://issues.apache.org/jira/browse/IGNITE-20773")
+ public void test() throws Exception {
+ TableImpl tbl = (TableImpl) node(0).tables().table(TABLE_NAME);
+
+ var tblReplicationGrp = new TablePartitionId(tbl.tableId(), 0);
+
+ CompletableFuture<ReplicaMeta> primaryReplicaFut = node(0).placementDriver().awaitPrimaryReplica(
+ tblReplicationGrp,
+ node(0).clock().now(),
+ 10,
+ SECONDS
+ );
+
+ assertThat(primaryReplicaFut, willCompleteSuccessfully());
+
+ String leaseholder = primaryReplicaFut.join().getLeaseholder();
+
+ IgniteImpl commitPartNode = IntStream.range(0, initialNodes()).mapToObj(this::node).filter(n -> leaseholder.equals(n.name()))
+ .findFirst().get();
+
+ log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());
+
+ IgniteImpl txCrdNode = IntStream.range(1, initialNodes()).mapToObj(this::node).filter(n -> !leaseholder.equals(n.name()))
+ .findFirst().get();
+
+ log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());
+
+ UUID orphanTxId = startTransactionAndStopNode(txCrdNode);
+
+ CompletableFuture<UUID> recoveryTxMsgCaptureFut = new CompletableFuture<>();
+
+
Review Comment:
empty line
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.tx.impl;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The class detects transactions that are left without a coordinator but still hold locks. For that orphan transaction, the recovery
+ * message is sent to the commit partition replication group.
+ */
+public class OrphanDetector {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(OrphanDetector.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ private static final long AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC = 10;
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ private final PlacementDriver placementDriver;
+
+ private final HybridClock clock;
+
+ /** The local map for tx states. */
+ private ConcurrentHashMap<UUID, TxStateMeta> txStateMap;
+
+ /**
+ * The constructor.
+ *
+ * @param clusterService Cluster service.
+ * @param placementDriver Placement driver.
+ * @param clock Clock.
+ */
+ public OrphanDetector(ClusterService clusterService, PlacementDriver placementDriver, HybridClock clock) {
+ this.clusterService = clusterService;
+ this.placementDriver = placementDriver;
+ this.clock = clock;
+ }
+
+
+ /**
+ * Starts the detector.
+ *
+ * @param txStateMap Transaction state map.
+ */
+ public void start(ConcurrentHashMap<UUID, TxStateMeta> txStateMap) {
+ this.txStateMap = txStateMap;
+ // Subscribe to lock conflicts here.
+ }
+
+ /**
+ * Stops the detector.
+ */
+ public void stop() {
+ busyLock.block();
+ // Unsubscribe from lock conflicts here.
+ }
+
+ /**
+ * Sends {@link TxRecoveryMessage} if the transaction is orphaned.
+ *
+ * @param txId Transaction id that holds a lock.
+ * @return Future to complete.
+ */
+ private CompletableFuture<Void> handleLockHolder(UUID txId) {
Review Comment:
Again, private method with no usages. Is it supposed to be called on lock conflict?
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.tx.impl;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The class detects transactions that are left without a coordinator but still hold locks. For that orphan transaction, the recovery
+ * message is sent to the commit partition replication group.
+ */
+public class OrphanDetector {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(OrphanDetector.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ private static final long AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC = 10;
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ private final PlacementDriver placementDriver;
+
+ private final HybridClock clock;
+
+ /** The local map for tx states. */
+ private ConcurrentHashMap<UUID, TxStateMeta> txStateMap;
+
+ /**
+ * The constructor.
+ *
+ * @param clusterService Cluster service.
+ * @param placementDriver Placement driver.
+ * @param clock Clock.
+ */
+ public OrphanDetector(ClusterService clusterService, PlacementDriver placementDriver, HybridClock clock) {
+ this.clusterService = clusterService;
+ this.placementDriver = placementDriver;
+ this.clock = clock;
+ }
+
+
Review Comment:
empty line
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.tx.impl;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The class detects transactions that are left without a coordinator but still hold locks. For that orphan transaction, the recovery
+ * message is sent to the commit partition replication group.
+ */
+public class OrphanDetector {
Review Comment:
Not sure that this reflects the essence of what this class is doing. Actually it initiates the recovery process, it doesn't detect anything. But still, it's not clear at all, because there are no usages of this class. Is it supposed to be triggered by the lock manager on lock conflict? Seems that this part was forgotten.
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryProcessor.java:
##########
@@ -0,0 +1,109 @@
+/*
+ * 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.tx.impl;
+
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The processor receives transaction recovery messages. The recovery message is received when the node has a commit partition for the
+ * specific transaction.
+ */
+public class TxRecoveryProcessor {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryProcessor.class);
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The local map for tx states. */
+ private ConcurrentHashMap<UUID, TxStateMeta> txStateMap;
+
+ /**
+ * The constructor.
+ *
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryProcessor(ClusterService clusterService) {
+ this.clusterService = clusterService;
+ }
+
+ /**
+ * Starts the recovery processor.
+ *
+ * @param txStateMap Transaction state map.
+ */
+ public void start(ConcurrentHashMap<UUID, TxStateMeta> txStateMap) {
+ this.txStateMap = txStateMap;
+
+ clusterService.messagingService().addMessageHandler(TxMessageGroup.class, (msg, sender, correlationId) -> {
+ if (msg instanceof TxRecoveryMessage) {
+ if (busyLock.enterBusy()) {
+ try {
+ processTxRecoveryMessage((TxRecoveryMessage) msg);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ } else {
+ LOG.info("Transaction recovery message was ignored [msg={}]", msg);
+ }
+ }
+ });
+ }
+
+ /**
+ * Processes the recovery transaction message.
+ *
+ * @param msg Transaction recovery message.
+ */
+ private void processTxRecoveryMessage(TxRecoveryMessage msg) {
+ TxStateMeta txState = txStateMap.get(msg.txId());
+
+ if (txState == null || !TxState.isFinalState(txState.txState())) {
+ if (txState != null && clusterService.topologyService().getById(txState.txCoordinatorId()) == null) {
Review Comment:
Why the liveness of the coordinator should relate the txState?
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryProcessor.java:
##########
@@ -0,0 +1,109 @@
+/*
+ * 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.tx.impl;
+
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The processor receives transaction recovery messages. The recovery message is received when the node has a commit partition for the
+ * specific transaction.
+ */
+public class TxRecoveryProcessor {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryProcessor.class);
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The local map for tx states. */
+ private ConcurrentHashMap<UUID, TxStateMeta> txStateMap;
+
+ /**
+ * The constructor.
+ *
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryProcessor(ClusterService clusterService) {
+ this.clusterService = clusterService;
+ }
+
+ /**
+ * Starts the recovery processor.
+ *
+ * @param txStateMap Transaction state map.
+ */
+ public void start(ConcurrentHashMap<UUID, TxStateMeta> txStateMap) {
+ this.txStateMap = txStateMap;
+
+ clusterService.messagingService().addMessageHandler(TxMessageGroup.class, (msg, sender, correlationId) -> {
+ if (msg instanceof TxRecoveryMessage) {
+ if (busyLock.enterBusy()) {
+ try {
+ processTxRecoveryMessage((TxRecoveryMessage) msg);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ } else {
+ LOG.info("Transaction recovery message was ignored [msg={}]", msg);
+ }
+ }
+ });
+ }
+
+ /**
+ * Processes the recovery transaction message.
+ *
+ * @param msg Transaction recovery message.
+ */
+ private void processTxRecoveryMessage(TxRecoveryMessage msg) {
+ TxStateMeta txState = txStateMap.get(msg.txId());
+
+ if (txState == null || !TxState.isFinalState(txState.txState())) {
+ if (txState != null && clusterService.topologyService().getById(txState.txCoordinatorId()) == null) {
+ LOG.warn(
+ "The transaction coordinator is alive, which is why the recovery message is ignored [tx={}, crd={}].",
Review Comment:
Seems that this is inconsistent with the condition: `clusterService.topologyService().getById(txState.txCoordinatorId()) == null`.
Also, I would rephrase the message: `The transaction coordinator is alive, so the recovery message is ignored`
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java:
##########
@@ -590,7 +585,16 @@ public int pending() {
@Override
public void start() {
- replicaService.messagingService().addMessageHandler(ReplicaMessageGroup.class, this);
+ localNodeId = clusterService.topologyService().localMember().id();
+ clusterService.messagingService().addMessageHandler(ReplicaMessageGroup.class, this);
Review Comment:
What is this? Does it fix some bug? I see that `onReceived` was added before but never used
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1392532908
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMeta.java:
##########
@@ -30,22 +33,28 @@ public class TxStateMeta implements TransactionMeta {
private final String txCoordinatorId;
+ /** Identifier of the replication group that manages a transaction state. */
+ private ReplicationGroupId commitPartitionId;
Review Comment:
may be final
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1397320765
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -478,10 +479,19 @@ private CompletableFuture<?> processRequest(ReplicaRequest request, @Nullable Bo
// Saving state is not needed for full transactions.
if (!req.full()) {
- txManager.updateTxMeta(req.transactionId(), old -> new TxStateMeta(PENDING, senderId, null));
+ txManager.updateTxMeta(req.transactionId(), old -> new TxStateMeta(
+ PENDING,
+ senderId,
+ req.commitPartitionId().asTablePartitionId(),
+ null
+ ));
}
}
+ if (request instanceof TxRecoveryMessage) {
+ processTxRecoveryAction((TxRecoveryMessage) request);
Review Comment:
I agree.
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1399008240
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * 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.tx.impl;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.tx.TxState.isFinalState;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.replicator.ReplicaService;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.TopologyService;
+
+/**
+ * The class detects transactions that are left without a coordinator but still hold locks. For that orphan transaction, the recovery
+ * message is sent to the commit partition replication group.
+ */
+public class OrphanDetector {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(OrphanDetector.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ private static final long AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC = 10;
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Topology service. */
+ private final TopologyService topologyService;
+
+ /** Replica service. */
+ private final ReplicaService replicaService;
+
+ /** Placement driver. */
+ private final PlacementDriver placementDriver;
+
+ // TODO: IGNITE-20773 Uncomment this during implementation.
+ ///** Lock manager. */
+ //private final LockManager lockManager;
+
+ /** Hybrid clock. */
+ private final HybridClock clock;
+
+ /** Local transaction state storage. */
+ private Function<UUID, TxStateMeta> txLocalStateStorage;
+
+ /**
+ * The constructor.
+ *
+ * @param topologyService Topology service.
+ * @param replicaService Replica service.
+ * @param placementDriver Placement driver.
+ * @param clock Clock.
+ */
+ public OrphanDetector(
+ TopologyService topologyService,
+ ReplicaService replicaService,
+ PlacementDriver placementDriver,
+ //LockManager lockManager,
+ HybridClock clock) {
+ this.topologyService = topologyService;
+ this.replicaService = replicaService;
+ this.placementDriver = placementDriver;
+ //this.lockManager = lockManager;
+ this.clock = clock;
+ }
+
+ /**
+ * Starts the detector.
+ *
+ * @param txLocalStateStorage Local transaction state storage.
+ */
+ public void start(Function<UUID, TxStateMeta> txLocalStateStorage) {
+ this.txLocalStateStorage = txLocalStateStorage;
+ // TODO: IGNITE-20773 Subscribe to lock conflicts here.
+ }
+
+ /**
+ * Stops the detector.
+ */
+ public void stop() {
+ busyLock.block();
+ // TODO: IGNITE-20773 Unsubscribe from lock conflicts here.
+ }
+
+ /**
+ * Sends {@link TxRecoveryMessage} if the transaction is orphaned.
+ * TODO: IGNITE-20773 Invoke the method when the lock conflict is noted.
+ *
+ * @param txId Transaction id that holds a lock.
+ * @return Future to complete.
+ */
+ private CompletableFuture<Void> handleLockHolder(UUID txId) {
+ if (busyLock.enterBusy()) {
+ try {
+ return handleLockHolderInternal(txId);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ return failedFuture(new NodeStoppingException());
+ }
+
+ /**
+ * Sends {@link TxRecoveryMessage} if the transaction is orphaned.
+ *
+ * @param txId Transaction id that holds a lock.
+ * @return Future to complete.
+ */
+ private CompletableFuture<Void> handleLockHolderInternal(UUID txId) {
+ TxStateMeta txState = txLocalStateStorage.apply(txId);
+
+ assert txState != null : "The transaction is undefined in the local node [txId=" + txId + "].";
+
+ if (txState.txState() == TxState.ABANDONED
Review Comment:
This is a first attempt to analyze ABENDEND; the full analysis will be done in the next ticket.
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1392558279
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryProcessor.java:
##########
@@ -0,0 +1,109 @@
+/*
+ * 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.tx.impl;
+
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The processor receives transaction recovery messages. The recovery message is received when the node has a commit partition for the
+ * specific transaction.
+ */
+public class TxRecoveryProcessor {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryProcessor.class);
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The local map for tx states. */
+ private ConcurrentHashMap<UUID, TxStateMeta> txStateMap;
+
+ /**
+ * The constructor.
+ *
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryProcessor(ClusterService clusterService) {
+ this.clusterService = clusterService;
+ }
+
+ /**
+ * Starts the recovery processor.
+ *
+ * @param txStateMap Transaction state map.
+ */
+ public void start(ConcurrentHashMap<UUID, TxStateMeta> txStateMap) {
+ this.txStateMap = txStateMap;
+
+ clusterService.messagingService().addMessageHandler(TxMessageGroup.class, (msg, sender, correlationId) -> {
+ if (msg instanceof TxRecoveryMessage) {
+ if (busyLock.enterBusy()) {
+ try {
+ processTxRecoveryMessage((TxRecoveryMessage) msg);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ } else {
+ LOG.info("Transaction recovery message was ignored [msg={}]", msg);
+ }
+ }
+ });
+ }
+
+ /**
+ * Processes the recovery transaction message.
+ *
+ * @param msg Transaction recovery message.
+ */
+ private void processTxRecoveryMessage(TxRecoveryMessage msg) {
Review Comment:
Please help me to recollect the agreed process of handling the tx recovery message.
I thought the handler of this message was supposed to be the primary replica of the commit partition, but this code is executed on every node.
What am I missing 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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1392523375
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java:
##########
@@ -0,0 +1,152 @@
+/*
+ * 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.tx.impl;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The class detects transactions that are left without a coordinator but still hold locks. For that orphan transaction, the recovery
+ * message is sent to the commit partition replication group.
+ */
+public class OrphanDetector {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(OrphanDetector.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ private static final long AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC = 10;
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ private final PlacementDriver placementDriver;
+
+ private final HybridClock clock;
+
+ /** The local map for tx states. */
+ private ConcurrentHashMap<UUID, TxStateMeta> txStateMap;
+
+ /**
+ * The constructor.
+ *
+ * @param clusterService Cluster service.
+ * @param placementDriver Placement driver.
+ * @param clock Clock.
+ */
+ public OrphanDetector(ClusterService clusterService, PlacementDriver placementDriver, HybridClock clock) {
+ this.clusterService = clusterService;
+ this.placementDriver = placementDriver;
+ this.clock = clock;
+ }
+
+
+ /**
+ * Starts the detector.
+ *
+ * @param txStateMap Transaction state map.
+ */
+ public void start(ConcurrentHashMap<UUID, TxStateMeta> txStateMap) {
+ this.txStateMap = txStateMap;
+ // Subscribe to lock conflicts here.
+ }
Review Comment:
The internal map leaks into this method and `TxRecoveryProcessor.start` from TxManagerImpl. A better solution would be to use something like a Function<UUID, TxStateMeta> or a specific interface to keep the state encapsulated.
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1397187683
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -478,10 +479,19 @@ private CompletableFuture<?> processRequest(ReplicaRequest request, @Nullable Bo
// Saving state is not needed for full transactions.
if (!req.full()) {
- txManager.updateTxMeta(req.transactionId(), old -> new TxStateMeta(PENDING, senderId, null));
+ txManager.updateTxMeta(req.transactionId(), old -> new TxStateMeta(
+ PENDING,
+ senderId,
+ req.commitPartitionId().asTablePartitionId(),
+ null
+ ));
}
}
+ if (request instanceof TxRecoveryMessage) {
+ processTxRecoveryAction((TxRecoveryMessage) request);
Review Comment:
do we need to proceed executing this request after processTxRecoveryAction? Looks like we can add a `return` statement 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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1395953179
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java:
##########
@@ -590,7 +585,16 @@ public int pending() {
@Override
public void start() {
- replicaService.messagingService().addMessageHandler(ReplicaMessageGroup.class, this);
+ localNodeId = clusterService.topologyService().localMember().id();
+ clusterService.messagingService().addMessageHandler(ReplicaMessageGroup.class, this);
Review Comment:
I don't think that it is not used. The method is used for handling replica requests.
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1395961973
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/HeapLockManager.java:
##########
@@ -167,7 +167,7 @@ public Waiter waiter(LockKey key, UUID txId) {
/**
* A lock state.
*/
- private static class LockState {
Review Comment:
Reverted
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "denis-chudov (via GitHub)" <gi...@apache.org>.
denis-chudov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1397338765
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java:
##########
@@ -63,8 +62,8 @@ public class OrphanDetector {
/** Placement driver. */
private final PlacementDriver placementDriver;
- /** Lock manager. */
- private final LockManager lockManager;
+ ///** Lock manager. */
+ //private final LockManager lockManager;
Review Comment:
Seems that TODO is needed here
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -490,6 +500,30 @@ private CompletableFuture<?> processRequest(ReplicaRequest request, @Nullable Bo
.thenCompose(opStartTimestamp -> processOperationRequest(request, isPrimary, senderId, opTsIfDirectRo));
}
+ /**
+ * Processes transaction recovery request.
Review Comment:
```suggestion
* Processes transaction recovery request on a commit partition.
```
##########
modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java:
##########
@@ -490,6 +500,30 @@ private CompletableFuture<?> processRequest(ReplicaRequest request, @Nullable Bo
.thenCompose(opStartTimestamp -> processOperationRequest(request, isPrimary, senderId, opTsIfDirectRo));
}
+ /**
+ * Processes transaction recovery request.
+ *
+ * @param request Tx recovery request.
+ * @return The future is complete when the transaction state is finalized.
+ */
+ private CompletableFuture<Void> processTxRecoveryAction(TxRecoveryMessage request) {
+ UUID txId = request.txId();
+
+ TxMeta txMeta = txStateStorage.get(txId);
+
+ // Check whether a transaction has already been finished.
+ boolean transactionAlreadyFinished = txMeta != null && isFinalState(txMeta.txState());
+
+ if (transactionAlreadyFinished) {
+ return completedFuture(null);
+ }
+
+ LOG.info("Orphan transactions have to be aborted [tx={}].", txId);
Review Comment:
```suggestion
LOG.info("Orphan transaction has to be aborted [tx={}].", txId);
```
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java:
##########
@@ -349,9 +332,17 @@ public CompletableFuture<Void> finish(
// than all the read timestamps processed before.
// Every concurrent operation will now use a finish future from the finishing state meta and get only final transaction
// state after the transaction is finished.
- TxStateMetaFinishing finishingStateMeta = new TxStateMetaFinishing(coordinatorId());
+ AtomicReference<TxStateMetaFinishing> finishingStateMetaRef = new AtomicReference<>();
+
+ updateTxMeta(txId, old -> {
+ var finishingState = new TxStateMetaFinishing(localNodeId, old == null ? null : old.commitPartitionId());
- updateTxMeta(txId, old -> finishingStateMeta);
+ finishingStateMetaRef.set(finishingState);
+
+ return finishingState;
+ });
Review Comment:
It is - this will allow you to get rid of `finishingStateMetaRef`
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java:
##########
@@ -0,0 +1,179 @@
+/*
+ * 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.tx.impl;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.replicator.ReplicaService;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.TopologyService;
+
+/**
+ * The class detects transactions that are left without a coordinator but still hold locks. For that orphan transaction, the recovery
+ * message is sent to the commit partition replication group.
+ */
+public class OrphanDetector {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(OrphanDetector.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ private static final long AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC = 10;
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Topology service. */
+ private final TopologyService topologyService;
+
+ /** Replica service. */
+ private final ReplicaService replicaService;
+
+ /** Placement driver. */
+ private final PlacementDriver placementDriver;
+
+ ///** Lock manager. */
+ //private final LockManager lockManager;
+
+ /** Hybrid clock. */
+ private final HybridClock clock;
+
+ /** Local transaction state storage. */
+ private Function<UUID, TxStateMeta> txLocalStateStorage;
+
+ /**
+ * The constructor.
+ *
+ * @param topologyService Topology service.
+ * @param replicaService Replica service.
+ * @param placementDriver Placement driver.
+ * @param clock Clock.
+ */
+ public OrphanDetector(
+ TopologyService topologyService,
+ ReplicaService replicaService,
+ PlacementDriver placementDriver,
+ //LockManager lockManager,
+ HybridClock clock) {
+ this.topologyService = topologyService;
+ this.replicaService = replicaService;
+ this.placementDriver = placementDriver;
+ //this.lockManager = lockManager;
+ this.clock = clock;
+ }
+
+ /**
+ * Starts the detector.
+ *
+ * @param txLocalStateStorage Local transaction state storage.
+ */
+ public void start(Function<UUID, TxStateMeta> txLocalStateStorage) {
+ this.txLocalStateStorage = txLocalStateStorage;
+ // TODO: IGNITE-20773 Subscribe to lock conflicts here.
+ }
+
+ /**
+ * Stops the detector.
+ */
+ public void stop() {
+ busyLock.block();
+ // TODO: IGNITE-20773 Unsubscribe from lock conflicts here.
+ }
+
+ /**
+ * Sends {@link TxRecoveryMessage} if the transaction is orphaned.
+ * TODO: IGNITE-20773 Invoke the method when the lock conflict is noted.
+ *
+ * @param txId Transaction id that holds a lock.
+ * @return Future to complete.
+ */
+ private CompletableFuture<Void> handleLockHolder(UUID txId) {
+ if (busyLock.enterBusy()) {
+ try {
+ return handleLockHolderInternal(txId);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ return failedFuture(new NodeStoppingException());
+ }
+
+ /**
+ * Sends {@link TxRecoveryMessage} if the transaction is orphaned.
+ *
+ * @param txId Transaction id that holds a lock.
+ * @return Future to complete.
+ */
+ private CompletableFuture<Void> handleLockHolderInternal(UUID txId) {
+ TxStateMeta txState = txLocalStateStorage.apply(txId);
+
+ assert txState != null : "The transaction is undefined in the local node [txId=" + txId + "].";
+
+ if (topologyService.getById(txState.txCoordinatorId()) == null) {
+ LOG.info(
+ "Conflict was found, and the coordinator of the transaction that holds a lock is not available "
+ + "[txId={}, txCrd={}].",
+ txId,
+ txState.txCoordinatorId()
+ );
+
+ return placementDriver.awaitPrimaryReplica(
+ txState.commitPartitionId(),
+ clock.now(),
+ AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC,
+ SECONDS
+ ).thenCompose(replicaMeta -> {
+ ClusterNode commitPartPrimaryNode = topologyService.getByConsistentId(replicaMeta.getLeaseholder());
+
+ if (commitPartPrimaryNode == null) {
+ LOG.warn(
+ "The primary replica of the commit partition is not available [commitPartGrp={}, tx={}]",
+ txState.commitPartitionId(),
+ txId
+ );
+
+ return completedFuture(null);
Review Comment:
Failed future should be returned from here, it will allow us to mark this tx as abandoned (or can we do it right 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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov merged PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1398833472
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/OrphanDetector.java:
##########
@@ -0,0 +1,184 @@
+/*
+ * 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.tx.impl;
+
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
+import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.apache.ignite.internal.tx.TxState.isFinalState;
+
+import java.util.UUID;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Function;
+import org.apache.ignite.internal.hlc.HybridClock;
+import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.placementdriver.PlacementDriver;
+import org.apache.ignite.internal.replicator.ReplicaService;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.TopologyService;
+
+/**
+ * The class detects transactions that are left without a coordinator but still hold locks. For that orphan transaction, the recovery
+ * message is sent to the commit partition replication group.
+ */
+public class OrphanDetector {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(OrphanDetector.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ private static final long AWAIT_PRIMARY_REPLICA_TIMEOUT_SEC = 10;
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Topology service. */
+ private final TopologyService topologyService;
+
+ /** Replica service. */
+ private final ReplicaService replicaService;
+
+ /** Placement driver. */
+ private final PlacementDriver placementDriver;
+
+ // TODO: IGNITE-20773 Uncomment this during implementation.
+ ///** Lock manager. */
+ //private final LockManager lockManager;
+
+ /** Hybrid clock. */
+ private final HybridClock clock;
+
+ /** Local transaction state storage. */
+ private Function<UUID, TxStateMeta> txLocalStateStorage;
+
+ /**
+ * The constructor.
+ *
+ * @param topologyService Topology service.
+ * @param replicaService Replica service.
+ * @param placementDriver Placement driver.
+ * @param clock Clock.
+ */
+ public OrphanDetector(
+ TopologyService topologyService,
+ ReplicaService replicaService,
+ PlacementDriver placementDriver,
+ //LockManager lockManager,
+ HybridClock clock) {
+ this.topologyService = topologyService;
+ this.replicaService = replicaService;
+ this.placementDriver = placementDriver;
+ //this.lockManager = lockManager;
+ this.clock = clock;
+ }
+
+ /**
+ * Starts the detector.
+ *
+ * @param txLocalStateStorage Local transaction state storage.
+ */
+ public void start(Function<UUID, TxStateMeta> txLocalStateStorage) {
+ this.txLocalStateStorage = txLocalStateStorage;
+ // TODO: IGNITE-20773 Subscribe to lock conflicts here.
+ }
+
+ /**
+ * Stops the detector.
+ */
+ public void stop() {
+ busyLock.block();
+ // TODO: IGNITE-20773 Unsubscribe from lock conflicts here.
+ }
+
+ /**
+ * Sends {@link TxRecoveryMessage} if the transaction is orphaned.
+ * TODO: IGNITE-20773 Invoke the method when the lock conflict is noted.
+ *
+ * @param txId Transaction id that holds a lock.
+ * @return Future to complete.
+ */
+ private CompletableFuture<Void> handleLockHolder(UUID txId) {
+ if (busyLock.enterBusy()) {
+ try {
+ return handleLockHolderInternal(txId);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
+ return failedFuture(new NodeStoppingException());
+ }
+
+ /**
+ * Sends {@link TxRecoveryMessage} if the transaction is orphaned.
+ *
+ * @param txId Transaction id that holds a lock.
+ * @return Future to complete.
+ */
+ private CompletableFuture<Void> handleLockHolderInternal(UUID txId) {
+ TxStateMeta txState = txLocalStateStorage.apply(txId);
+
+ assert txState != null : "The transaction is undefined in the local node [txId=" + txId + "].";
+
+ if (txState.txState() == TxState.ABANDONED
Review Comment:
In what task we were going to tackle abandoned transactions?
##########
modules/table/src/testFixtures/java/org/apache/ignite/internal/table/impl/DummyInternalTableImpl.java:
##########
@@ -433,14 +437,26 @@ public TxManager txManager() {
* @param replicaSvc Replica service to use.
*/
public static TxManagerImpl txManager(ReplicaService replicaSvc) {
- return new TxManagerImpl(
+ TopologyService topologyService = mock(TopologyService.class);
+ when(topologyService.localMember()).thenReturn(LOCAL_NODE);
+
+ ClusterService clusterService = mock(ClusterService.class);
+ when(clusterService.messagingService()).thenReturn(mock(MessagingService.class));
+ when(clusterService.topologyService()).thenReturn(topologyService);
+
+ var txManager = new TxManagerImpl(
+ clusterService,
replicaSvc,
new HeapLockManager(),
CLOCK,
new TransactionIdGenerator(0xdeadbeef),
- LOCAL_NODE::id,
- TEST_PLACEMENT_DRIVER
+ TEST_PLACEMENT_DRIVER,
+ () -> DEFAULT_IDLE_SAFE_TIME_PROPAGATION_PERIOD_MILLISECONDS
);
+
+ txManager.start();
Review Comment:
Unfortunately nobody calls stop to match this start.
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1395946326
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java:
##########
@@ -349,9 +332,17 @@ public CompletableFuture<Void> finish(
// than all the read timestamps processed before.
// Every concurrent operation will now use a finish future from the finishing state meta and get only final transaction
// state after the transaction is finished.
- TxStateMetaFinishing finishingStateMeta = new TxStateMetaFinishing(coordinatorId());
+ AtomicReference<TxStateMetaFinishing> finishingStateMetaRef = new AtomicReference<>();
+
+ updateTxMeta(txId, old -> {
+ var finishingState = new TxStateMetaFinishing(localNodeId, old == null ? null : old.commitPartitionId());
- updateTxMeta(txId, old -> finishingStateMeta);
+ finishingStateMetaRef.set(finishingState);
+
+ return finishingState;
+ });
Review Comment:
It is not related to the patch.
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1395985746
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxRecoveryMessage.java:
##########
@@ -0,0 +1,35 @@
+/*
+ * 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.tx.message;
+
+import java.util.UUID;
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.annotations.Transferable;
+
+/**
+ * Transaction recovery message.
+ */
+@Transferable(TxMessageGroup.TX_RECOVERY_MSG)
+public interface TxRecoveryMessage extends NetworkMessage {
+ /**
+ * Gets a transaction id to resolve.
+ *
+ * @return Transaction id.
+ */
Review Comment:
For now, we are not sending a cleanup request anywhere. Before, we should agree on the node cleanup procedure for a node (currently, we have the procedure for a 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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1392531740
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxStateMetaFinishing.java:
##########
@@ -35,9 +36,10 @@ public class TxStateMetaFinishing extends TxStateMeta {
* Constructor.
*
* @param txCoordinatorId Transaction coordinator id.
+ * @param commitPartitionId Commit partition id.
*/
- public TxStateMetaFinishing(String txCoordinatorId) {
- super(TxState.FINISHING, txCoordinatorId, null);
+ public TxStateMetaFinishing(String txCoordinatorId, ReplicationGroupId commitPartitionId) {
Review Comment:
Please mark commitPartitionId as `@Nullable`
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1392565761
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryProcessor.java:
##########
@@ -0,0 +1,109 @@
+/*
+ * 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.tx.impl;
+
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.internal.util.IgniteSpinBusyLock;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The processor receives transaction recovery messages. The recovery message is received when the node has a commit partition for the
+ * specific transaction.
+ */
+public class TxRecoveryProcessor {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryProcessor.class);
+
+ /** Busy lock to stop synchronously. */
+ private final IgniteSpinBusyLock busyLock = new IgniteSpinBusyLock();
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The local map for tx states. */
+ private ConcurrentHashMap<UUID, TxStateMeta> txStateMap;
+
+ /**
+ * The constructor.
+ *
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryProcessor(ClusterService clusterService) {
+ this.clusterService = clusterService;
+ }
+
+ /**
+ * Starts the recovery processor.
+ *
+ * @param txStateMap Transaction state map.
+ */
+ public void start(ConcurrentHashMap<UUID, TxStateMeta> txStateMap) {
+ this.txStateMap = txStateMap;
+
+ clusterService.messagingService().addMessageHandler(TxMessageGroup.class, (msg, sender, correlationId) -> {
+ if (msg instanceof TxRecoveryMessage) {
+ if (busyLock.enterBusy()) {
+ try {
+ processTxRecoveryMessage((TxRecoveryMessage) msg);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ } else {
+ LOG.info("Transaction recovery message was ignored [msg={}]", msg);
+ }
+ }
+ });
+ }
+
+ /**
+ * Processes the recovery transaction message.
+ *
+ * @param msg Transaction recovery message.
+ */
+ private void processTxRecoveryMessage(TxRecoveryMessage msg) {
Review Comment:
Okay, I checked the code again and see that this message is sent onto the commit partition primary only from `OrphanDetector`. This approach looks different from the other messages handling (which is done in PartitionReplicaListener). Is there any specific reason we have this class?
--
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-20685 Implement ability to trigger transaction recovery [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #2832:
URL: https://github.com/apache/ignite-3/pull/2832#discussion_r1393830157
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyTransactionImpl.java:
##########
@@ -116,7 +116,7 @@ public CompletableFuture<Void> finish(boolean commit, HybridTimestamp executionT
return ((TxManagerImpl) txManager).completeReadOnlyTransactionFuture(new TxIdAndTimestamp(readTimestamp, id()))
.thenRun(() -> txManager.updateTxMeta(
id(),
- old -> new TxStateMeta(COMMITED, old.txCoordinatorId(), old.commitTimestamp())
+ old -> new TxStateMeta(COMMITED, old.txCoordinatorId(), null, old.commitTimestamp())
Review Comment:
I don't think that commit partitioning for RO transactions is required.
--
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