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/06 18:47:40 UTC
[PR] IGNITE-20685 Implement ability to trigger transaction recovery [ignite-3]
vldpyatkov opened a new pull request, #2806:
URL: https://github.com/apache/ignite-3/pull/2806
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 #2806:
URL: https://github.com/apache/ignite-3/pull/2806#discussion_r1385005976
##########
modules/runner/src/integrationTest/java/org/apache/ignite/internal/table/ItTransactionConflictTest.java:
##########
@@ -0,0 +1,160 @@
+/*
+ * 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.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
+ 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 coordinator is determined [node={}].", commitPartNode.name());
Review Comment:
Mistake here, it is a commit partiton node.
--
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 #2806:
URL: https://github.com/apache/ignite-3/pull/2806#discussion_r1386116784
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/event/LockEventParameters.java:
##########
@@ -0,0 +1,78 @@
+/*
+ * 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.event;
+
+import java.util.UUID;
+import org.apache.ignite.internal.event.EventParameters;
+import org.apache.ignite.internal.tx.LockKey;
+
+/**
+ * The class presents parameters that could be interesting to handle with an event.
Review Comment:
You might like the following description instead:
> This class encapsulates event parameters related to lock management.
--
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 closed pull request #2806: IGNITE-20685 Implement ability to trigger transaction recovery
URL: https://github.com/apache/ignite-3/pull/2806
--
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 #2806:
URL: https://github.com/apache/ignite-3/pull/2806#discussion_r1384987026
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryManager.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.concurrent.CompletableFuture;
+import org.apache.ignite.internal.event.EventListener;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.event.LockEvent;
+import org.apache.ignite.internal.tx.event.LockEventParameters;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The manager sends and receives transaction recovery messages.
+ * <ul>
+ * <li>The recovery messages are sent when a lock conflict is found and the transaction coordinator is not available.</li>
+ * <li>The recovery message is received when the node has a commit partition for the specific transaction.</li>
+ * </ul>
+ */
+public class TxRecoveryManager implements IgniteComponent {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryManager.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ /** Lock manager. */
+ private final LockManager lockManager;
+
+ /** Transaction manager. */
+ private final TxManager txManager;
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The listener tries to clarify the holder transaction state to avoid permanent locking. */
+ private final EventListener<LockEventParameters> lockConflictListener;
+
+ /**
+ * The constructor.
+ *
+ * @param lockManager Lock manager.
+ * @param txManager Transaction manager.
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryManager(
+ LockManager lockManager,
+ TxManager txManager,
+ ClusterService clusterService
+ ) {
+ this.lockManager = lockManager;
+ this.txManager = txManager;
+ this.clusterService = clusterService;
+
+ lockConflictListener = (parameters, exception) -> {
+ TxStateMeta txSate = txManager.stateMeta(parameters.getLockHolderTx());
+
+ if (txSate == null || clusterService.topologyService().getById(txSate.txCoordinatorId()) == null) {
+ LOG.info("Conflict was found, and the coordinator of the transaction that holds a lock is not available "
+ + "[lockHolderTx={}, lockAcquirerTx={}].",
+ parameters.getLockHolderTx(),
+ parameters.getLockAcquirerTx()
+ );
+
+ //TODO:IGNITE-20735 Obviously, the commit partition node does not have to be at the local node.
+ ClusterNode cmpPartNode = clusterService.topologyService().localMember();
+
+ clusterService.messagingService().weakSend(cmpPartNode, FACTORY.txRecoveryMessage()
+ .txId(parameters.getLockHolderTx())
+ .build());
+ }
+
+ return CompletableFuture.completedFuture(false);
+ };
+ }
+
+ @Override
+ public void start() {
+ lockManager.listen(LockEvent.CONFLICT_FOUND, lockConflictListener);
Review Comment:
I thnk a method reference could work fine here, no need to add the field `lockConflictListener`
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryManager.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.concurrent.CompletableFuture;
+import org.apache.ignite.internal.event.EventListener;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.event.LockEvent;
+import org.apache.ignite.internal.tx.event.LockEventParameters;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The manager sends and receives transaction recovery messages.
+ * <ul>
+ * <li>The recovery messages are sent when a lock conflict is found and the transaction coordinator is not available.</li>
+ * <li>The recovery message is received when the node has a commit partition for the specific transaction.</li>
+ * </ul>
+ */
+public class TxRecoveryManager implements IgniteComponent {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryManager.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ /** Lock manager. */
+ private final LockManager lockManager;
+
+ /** Transaction manager. */
+ private final TxManager txManager;
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The listener tries to clarify the holder transaction state to avoid permanent locking. */
+ private final EventListener<LockEventParameters> lockConflictListener;
+
+ /**
+ * The constructor.
+ *
+ * @param lockManager Lock manager.
+ * @param txManager Transaction manager.
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryManager(
+ LockManager lockManager,
+ TxManager txManager,
+ ClusterService clusterService
+ ) {
+ this.lockManager = lockManager;
+ this.txManager = txManager;
+ this.clusterService = clusterService;
+
+ lockConflictListener = (parameters, exception) -> {
+ TxStateMeta txSate = txManager.stateMeta(parameters.getLockHolderTx());
+
+ if (txSate == null || clusterService.topologyService().getById(txSate.txCoordinatorId()) == null) {
+ LOG.info("Conflict was found, and the coordinator of the transaction that holds a lock is not available "
+ + "[lockHolderTx={}, lockAcquirerTx={}].",
+ parameters.getLockHolderTx(),
+ parameters.getLockAcquirerTx()
+ );
+
+ //TODO:IGNITE-20735 Obviously, the commit partition node does not have to be at the local node.
+ ClusterNode cmpPartNode = clusterService.topologyService().localMember();
+
+ clusterService.messagingService().weakSend(cmpPartNode, FACTORY.txRecoveryMessage()
+ .txId(parameters.getLockHolderTx())
+ .build());
+ }
+
+ return CompletableFuture.completedFuture(false);
+ };
+ }
+
+ @Override
+ public void start() {
+ lockManager.listen(LockEvent.CONFLICT_FOUND, lockConflictListener);
+
+ clusterService.messagingService().addMessageHandler(TxMessageGroup.class, (msg, sender, correlationId) -> {
+ if (msg instanceof TxRecoveryMessage) {
+ var txRecoveryMsg = (TxRecoveryMessage) msg;
+
+ TxStateMeta txState = txManager.stateMeta(txRecoveryMsg.txId());
+
+ if (txState == null || !TxState.isFinalState(txState.txState())) {
+ assert txState == null || clusterService.topologyService().getById(txState.txCoordinatorId()) == null :
+ "Transaction coordinator is alive [tx=" + txRecoveryMsg.txId() + ", crd=" + txState.txCoordinatorId() + ']';
+
+ LOG.warn("Transaction have to be aborted [tx={}].", txRecoveryMsg.txId());
Review Comment:
```suggestion
LOG.warn("Transaction has to be aborted [tx={}].", txRecoveryMsg.txId());
```
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryManager.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.concurrent.CompletableFuture;
+import org.apache.ignite.internal.event.EventListener;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.event.LockEvent;
+import org.apache.ignite.internal.tx.event.LockEventParameters;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The manager sends and receives transaction recovery messages.
+ * <ul>
+ * <li>The recovery messages are sent when a lock conflict is found and the transaction coordinator is not available.</li>
+ * <li>The recovery message is received when the node has a commit partition for the specific transaction.</li>
+ * </ul>
+ */
+public class TxRecoveryManager implements IgniteComponent {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryManager.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ /** Lock manager. */
+ private final LockManager lockManager;
+
+ /** Transaction manager. */
+ private final TxManager txManager;
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The listener tries to clarify the holder transaction state to avoid permanent locking. */
+ private final EventListener<LockEventParameters> lockConflictListener;
+
+ /**
+ * The constructor.
+ *
+ * @param lockManager Lock manager.
+ * @param txManager Transaction manager.
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryManager(
+ LockManager lockManager,
+ TxManager txManager,
+ ClusterService clusterService
+ ) {
+ this.lockManager = lockManager;
+ this.txManager = txManager;
+ this.clusterService = clusterService;
+
+ lockConflictListener = (parameters, exception) -> {
+ TxStateMeta txSate = txManager.stateMeta(parameters.getLockHolderTx());
+
+ if (txSate == null || clusterService.topologyService().getById(txSate.txCoordinatorId()) == null) {
+ LOG.info("Conflict was found, and the coordinator of the transaction that holds a lock is not available "
+ + "[lockHolderTx={}, lockAcquirerTx={}].",
Review Comment:
pls add coordinator id
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryManager.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.concurrent.CompletableFuture;
+import org.apache.ignite.internal.event.EventListener;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.event.LockEvent;
+import org.apache.ignite.internal.tx.event.LockEventParameters;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The manager sends and receives transaction recovery messages.
+ * <ul>
+ * <li>The recovery messages are sent when a lock conflict is found and the transaction coordinator is not available.</li>
+ * <li>The recovery message is received when the node has a commit partition for the specific transaction.</li>
+ * </ul>
+ */
+public class TxRecoveryManager implements IgniteComponent {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryManager.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ /** Lock manager. */
+ private final LockManager lockManager;
+
+ /** Transaction manager. */
+ private final TxManager txManager;
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The listener tries to clarify the holder transaction state to avoid permanent locking. */
+ private final EventListener<LockEventParameters> lockConflictListener;
+
+ /**
+ * The constructor.
+ *
+ * @param lockManager Lock manager.
+ * @param txManager Transaction manager.
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryManager(
+ LockManager lockManager,
+ TxManager txManager,
+ ClusterService clusterService
+ ) {
+ this.lockManager = lockManager;
+ this.txManager = txManager;
+ this.clusterService = clusterService;
+
+ lockConflictListener = (parameters, exception) -> {
+ TxStateMeta txSate = txManager.stateMeta(parameters.getLockHolderTx());
+
+ if (txSate == null || clusterService.topologyService().getById(txSate.txCoordinatorId()) == null) {
+ LOG.info("Conflict was found, and the coordinator of the transaction that holds a lock is not available "
+ + "[lockHolderTx={}, lockAcquirerTx={}].",
+ parameters.getLockHolderTx(),
+ parameters.getLockAcquirerTx()
+ );
+
+ //TODO:IGNITE-20735 Obviously, the commit partition node does not have to be at the local node.
Review Comment:
The ticket says:
`Transactions X that detects (detection logic will be covered in a separate ticket) that coordinator is dead awaits commitPartition primary replica and sends initiateRecoveryReplicaRequest to it in a fully asynchronous manner`
Why do you just pick a local node? IMO the part of sending the initiate recovery message should be fully covered by this PR. The mentioned ticket IGNITE-20735 hasn't anything in description about this part of work.
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryManager.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.concurrent.CompletableFuture;
+import org.apache.ignite.internal.event.EventListener;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.event.LockEvent;
+import org.apache.ignite.internal.tx.event.LockEventParameters;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The manager sends and receives transaction recovery messages.
+ * <ul>
+ * <li>The recovery messages are sent when a lock conflict is found and the transaction coordinator is not available.</li>
+ * <li>The recovery message is received when the node has a commit partition for the specific transaction.</li>
+ * </ul>
+ */
+public class TxRecoveryManager implements IgniteComponent {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryManager.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ /** Lock manager. */
+ private final LockManager lockManager;
+
+ /** Transaction manager. */
+ private final TxManager txManager;
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The listener tries to clarify the holder transaction state to avoid permanent locking. */
+ private final EventListener<LockEventParameters> lockConflictListener;
+
+ /**
+ * The constructor.
+ *
+ * @param lockManager Lock manager.
+ * @param txManager Transaction manager.
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryManager(
+ LockManager lockManager,
+ TxManager txManager,
+ ClusterService clusterService
+ ) {
+ this.lockManager = lockManager;
+ this.txManager = txManager;
+ this.clusterService = clusterService;
+
+ lockConflictListener = (parameters, exception) -> {
+ TxStateMeta txSate = txManager.stateMeta(parameters.getLockHolderTx());
+
+ if (txSate == null || clusterService.topologyService().getById(txSate.txCoordinatorId()) == null) {
+ LOG.info("Conflict was found, and the coordinator of the transaction that holds a lock is not available "
+ + "[lockHolderTx={}, lockAcquirerTx={}].",
+ parameters.getLockHolderTx(),
+ parameters.getLockAcquirerTx()
+ );
+
+ //TODO:IGNITE-20735 Obviously, the commit partition node does not have to be at the local node.
+ ClusterNode cmpPartNode = clusterService.topologyService().localMember();
Review Comment:
```suggestion
ClusterNode commitPartNode = clusterService.topologyService().localMember();
```
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryManager.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.concurrent.CompletableFuture;
+import org.apache.ignite.internal.event.EventListener;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.event.LockEvent;
+import org.apache.ignite.internal.tx.event.LockEventParameters;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The manager sends and receives transaction recovery messages.
+ * <ul>
+ * <li>The recovery messages are sent when a lock conflict is found and the transaction coordinator is not available.</li>
+ * <li>The recovery message is received when the node has a commit partition for the specific transaction.</li>
+ * </ul>
+ */
+public class TxRecoveryManager implements IgniteComponent {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryManager.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ /** Lock manager. */
+ private final LockManager lockManager;
+
+ /** Transaction manager. */
+ private final TxManager txManager;
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The listener tries to clarify the holder transaction state to avoid permanent locking. */
+ private final EventListener<LockEventParameters> lockConflictListener;
+
+ /**
+ * The constructor.
+ *
+ * @param lockManager Lock manager.
+ * @param txManager Transaction manager.
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryManager(
+ LockManager lockManager,
+ TxManager txManager,
+ ClusterService clusterService
+ ) {
+ this.lockManager = lockManager;
+ this.txManager = txManager;
+ this.clusterService = clusterService;
+
+ lockConflictListener = (parameters, exception) -> {
+ TxStateMeta txSate = txManager.stateMeta(parameters.getLockHolderTx());
Review Comment:
```suggestion
TxStateMeta txState = txManager.stateMeta(parameters.getLockHolderTx());
```
--
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 #2806:
URL: https://github.com/apache/ignite-3/pull/2806#discussion_r1385005654
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryManager.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.concurrent.CompletableFuture;
+import org.apache.ignite.internal.event.EventListener;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.event.LockEvent;
+import org.apache.ignite.internal.tx.event.LockEventParameters;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The manager sends and receives transaction recovery messages.
+ * <ul>
+ * <li>The recovery messages are sent when a lock conflict is found and the transaction coordinator is not available.</li>
+ * <li>The recovery message is received when the node has a commit partition for the specific transaction.</li>
+ * </ul>
+ */
+public class TxRecoveryManager implements IgniteComponent {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryManager.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ /** Lock manager. */
+ private final LockManager lockManager;
+
+ /** Transaction manager. */
+ private final TxManager txManager;
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The listener tries to clarify the holder transaction state to avoid permanent locking. */
+ private final EventListener<LockEventParameters> lockConflictListener;
+
+ /**
+ * The constructor.
+ *
+ * @param lockManager Lock manager.
+ * @param txManager Transaction manager.
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryManager(
+ LockManager lockManager,
+ TxManager txManager,
+ ClusterService clusterService
+ ) {
+ this.lockManager = lockManager;
+ this.txManager = txManager;
+ this.clusterService = clusterService;
+
+ lockConflictListener = (parameters, exception) -> {
+ TxStateMeta txSate = txManager.stateMeta(parameters.getLockHolderTx());
+
+ if (txSate == null || clusterService.topologyService().getById(txSate.txCoordinatorId()) == null) {
+ LOG.info("Conflict was found, and the coordinator of the transaction that holds a lock is not available "
+ + "[lockHolderTx={}, lockAcquirerTx={}].",
+ parameters.getLockHolderTx(),
+ parameters.getLockAcquirerTx()
+ );
+
+ //TODO:IGNITE-20735 Obviously, the commit partition node does not have to be at the local node.
Review Comment:
The ticket says:
`Transactions X that detects (detection logic will be covered in a separate ticket) that coordinator is dead awaits commitPartition primary replica and sends initiateRecoveryReplicaRequest to it in a fully asynchronous manner`
Why do you just pick a local node? IMO the part of sending the initiate recovery message should be fully covered by this PR. However it can be not trivial, then it requires separate ticket.
But the mentioned ticket IGNITE-20735 hasn't anything in description about this part of work.
--
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 #2806:
URL: https://github.com/apache/ignite-3/pull/2806#discussion_r1386101637
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryManager.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.concurrent.CompletableFuture;
+import org.apache.ignite.internal.event.EventListener;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.event.LockEvent;
+import org.apache.ignite.internal.tx.event.LockEventParameters;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The manager sends and receives transaction recovery messages.
+ * <ul>
+ * <li>The recovery messages are sent when a lock conflict is found and the transaction coordinator is not available.</li>
+ * <li>The recovery message is received when the node has a commit partition for the specific transaction.</li>
+ * </ul>
+ */
+public class TxRecoveryManager implements IgniteComponent {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryManager.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ /** Lock manager. */
+ private final LockManager lockManager;
+
+ /** Transaction manager. */
+ private final TxManager txManager;
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The listener tries to clarify the holder transaction state to avoid permanent locking. */
+ private final EventListener<LockEventParameters> lockConflictListener;
+
+ /**
+ * The constructor.
+ *
+ * @param lockManager Lock manager.
+ * @param txManager Transaction manager.
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryManager(
+ LockManager lockManager,
+ TxManager txManager,
+ ClusterService clusterService
+ ) {
+ this.lockManager = lockManager;
+ this.txManager = txManager;
+ this.clusterService = clusterService;
+
+ lockConflictListener = (parameters, exception) -> {
+ TxStateMeta txSate = txManager.stateMeta(parameters.getLockHolderTx());
+
+ if (txSate == null || clusterService.topologyService().getById(txSate.txCoordinatorId()) == null) {
+ LOG.info("Conflict was found, and the coordinator of the transaction that holds a lock is not available "
+ + "[lockHolderTx={}, lockAcquirerTx={}].",
+ parameters.getLockHolderTx(),
+ parameters.getLockAcquirerTx()
+ );
+
+ //TODO:IGNITE-20735 Obviously, the commit partition node does not have to be at the local node.
+ ClusterNode cmpPartNode = clusterService.topologyService().localMember();
+
+ clusterService.messagingService().weakSend(cmpPartNode, FACTORY.txRecoveryMessage()
+ .txId(parameters.getLockHolderTx())
+ .build());
+ }
+
+ return CompletableFuture.completedFuture(false);
+ };
+ }
+
+ @Override
+ public void start() {
+ lockManager.listen(LockEvent.CONFLICT_FOUND, lockConflictListener);
+
+ clusterService.messagingService().addMessageHandler(TxMessageGroup.class, (msg, sender, correlationId) -> {
+ if (msg instanceof TxRecoveryMessage) {
+ var txRecoveryMsg = (TxRecoveryMessage) msg;
+
+ TxStateMeta txState = txManager.stateMeta(txRecoveryMsg.txId());
+
+ if (txState == null || !TxState.isFinalState(txState.txState())) {
+ assert txState == null || clusterService.topologyService().getById(txState.txCoordinatorId()) == null :
Review Comment:
`TxRecoveryMessage` is supposed to be handled on a commit partition that is perfectly alive. Is there any need to throw an assertion error? How is the exception supposed to be handled?
--
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 #2806:
URL: https://github.com/apache/ignite-3/pull/2806#discussion_r1386101637
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxRecoveryManager.java:
##########
@@ -0,0 +1,130 @@
+/*
+ * 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.concurrent.CompletableFuture;
+import org.apache.ignite.internal.event.EventListener;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.tx.LockManager;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.internal.tx.TxState;
+import org.apache.ignite.internal.tx.TxStateMeta;
+import org.apache.ignite.internal.tx.event.LockEvent;
+import org.apache.ignite.internal.tx.event.LockEventParameters;
+import org.apache.ignite.internal.tx.message.TxMessageGroup;
+import org.apache.ignite.internal.tx.message.TxMessagesFactory;
+import org.apache.ignite.internal.tx.message.TxRecoveryMessage;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+/**
+ * The manager sends and receives transaction recovery messages.
+ * <ul>
+ * <li>The recovery messages are sent when a lock conflict is found and the transaction coordinator is not available.</li>
+ * <li>The recovery message is received when the node has a commit partition for the specific transaction.</li>
+ * </ul>
+ */
+public class TxRecoveryManager implements IgniteComponent {
+ /** The logger. */
+ private static final IgniteLogger LOG = Loggers.forClass(TxRecoveryManager.class);
+
+ /** Tx messages factory. */
+ private static final TxMessagesFactory FACTORY = new TxMessagesFactory();
+
+ /** Lock manager. */
+ private final LockManager lockManager;
+
+ /** Transaction manager. */
+ private final TxManager txManager;
+
+ /** Cluster service. */
+ private final ClusterService clusterService;
+
+ /** The listener tries to clarify the holder transaction state to avoid permanent locking. */
+ private final EventListener<LockEventParameters> lockConflictListener;
+
+ /**
+ * The constructor.
+ *
+ * @param lockManager Lock manager.
+ * @param txManager Transaction manager.
+ * @param clusterService Cluster service.
+ */
+ public TxRecoveryManager(
+ LockManager lockManager,
+ TxManager txManager,
+ ClusterService clusterService
+ ) {
+ this.lockManager = lockManager;
+ this.txManager = txManager;
+ this.clusterService = clusterService;
+
+ lockConflictListener = (parameters, exception) -> {
+ TxStateMeta txSate = txManager.stateMeta(parameters.getLockHolderTx());
+
+ if (txSate == null || clusterService.topologyService().getById(txSate.txCoordinatorId()) == null) {
+ LOG.info("Conflict was found, and the coordinator of the transaction that holds a lock is not available "
+ + "[lockHolderTx={}, lockAcquirerTx={}].",
+ parameters.getLockHolderTx(),
+ parameters.getLockAcquirerTx()
+ );
+
+ //TODO:IGNITE-20735 Obviously, the commit partition node does not have to be at the local node.
+ ClusterNode cmpPartNode = clusterService.topologyService().localMember();
+
+ clusterService.messagingService().weakSend(cmpPartNode, FACTORY.txRecoveryMessage()
+ .txId(parameters.getLockHolderTx())
+ .build());
+ }
+
+ return CompletableFuture.completedFuture(false);
+ };
+ }
+
+ @Override
+ public void start() {
+ lockManager.listen(LockEvent.CONFLICT_FOUND, lockConflictListener);
+
+ clusterService.messagingService().addMessageHandler(TxMessageGroup.class, (msg, sender, correlationId) -> {
+ if (msg instanceof TxRecoveryMessage) {
+ var txRecoveryMsg = (TxRecoveryMessage) msg;
+
+ TxStateMeta txState = txManager.stateMeta(txRecoveryMsg.txId());
+
+ if (txState == null || !TxState.isFinalState(txState.txState())) {
+ assert txState == null || clusterService.topologyService().getById(txState.txCoordinatorId()) == null :
Review Comment:
`TxRecoveryMessage` is handled on a commit partition that is perfectly alive. Is there any need to throw an assertion error? How is the exception supposed to be handled?
--
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