You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@ignite.apache.org by "Cyrill (via GitHub)" <gi...@apache.org> on 2024/02/06 10:44:07 UTC
[PR] IGNITE-21248 Add abandoned locks handling to HeapUnboundedLockManager [ignite-3]
Cyrill opened a new pull request, #3161:
URL: https://github.com/apache/ignite-3/pull/3161
https://issues.apache.org/jira/browse/IGNITE-21248
--
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-21248 Add abandoned locks handling to HeapUnboundedLockManager [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov merged PR #3161:
URL: https://github.com/apache/ignite-3/pull/3161
--
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-21248 Add abandoned locks handling to HeapUnboundedLockManager [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #3161:
URL: https://github.com/apache/ignite-3/pull/3161#discussion_r1481000020
##########
modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItTransactionRecoveryTest.java:
##########
@@ -771,6 +784,153 @@ public void testPrimaryFailureWhileInflightInProgressAfterFirstResponse() throws
assertNull(rec);
}
+ @Test
+ public void testTsRecoveryForCursor() throws Exception {
+ TableImpl tbl = (TableImpl) node(0).tables().table(TABLE_NAME);
+
+ RecordView view1 = node(0).tables().table(TABLE_NAME).recordView();
+
+ for (int i = 0; i < 10; i++) {
+ view1.upsert(null, Tuple.create().set("key", i).set("val", "preload"));
+ }
+
+ var tblReplicationGrp = new TablePartitionId(tbl.tableId(), 0);
+
+ String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+
+ IgniteImpl commitPartNode = commitPartitionPrimaryNode(leaseholder);
+
+ log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());
+
+ IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);
+
+ log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());
+
+ UUID orphanTxId = startTransactionWithCursorAndStopNode(txCrdNode);
+
+ IgniteImpl newCoordNode = node(0);
+
+ log.info("New transaction coordinator is chosen [node={}].", newCoordNode.name());
+
+ CompletableFuture<Void> txMsgCaptureFut = new CompletableFuture<>();
+
+ commitPartNode.dropMessages((nodeName, msg) -> {
+ if (msg instanceof TxRecoveryMessage) {
+ txMsgCaptureFut.complete(null);
+ }
+
+ return false;
+ });
+
+ Transaction tx = newCoordNode.transactions().begin();
+
+ RecordView view = newCoordNode.tables().table(TABLE_NAME).recordView();
+
+ var opFut = view.upsertAsync(tx, Tuple.create().set("key", 42).set("val", "new"));
+
+ try {
+ opFut.get();
+ } catch (Exception ex) {
+ log.info("Expected conflict that have to start recovery: " + ex.getMessage());
+ }
+
+ assertThat(txMsgCaptureFut, willCompleteSuccessfully());
+ }
+
+ private UUID startTransactionWithCursorAndStopNode(IgniteImpl txCrdNode) throws Exception {
+ InternalTransaction rwTx = (InternalTransaction) txCrdNode.transactions().begin();
+
+ scanSingleEntryAndLeaveCursorOpen((TableViewInternal) txCrdNode.tables().table(TABLE_NAME), rwTx, null, null);
+
+ String txCrdNodeId = txCrdNode.id();
+
+ txCrdNode.stop();
+
+ assertTrue(waitForCondition(
+ () -> node(0).clusterNodes().stream().filter(n -> txCrdNodeId.equals(n.id())).count() == 0,
+ 10_000)
+ );
+
+ return rwTx.id();
+ }
+
+ /**
+ * Starts a scan procedure for a specific transaction and reads only the first line from the cursor.
+ *
+ * @param tbl Scanned table.
+ * @param tx Transaction.
+ * @param idxId Index id.
+ * @throws Exception If failed.
+ */
+ private void scanSingleEntryAndLeaveCursorOpen(TableViewInternal tbl, InternalTransaction tx, Integer idxId, BinaryTuple exactKey)
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-21248 Add abandoned locks handling to HeapUnboundedLockManager [ignite-3]
Posted by "vldpyatkov (via GitHub)" <gi...@apache.org>.
vldpyatkov commented on code in PR #3161:
URL: https://github.com/apache/ignite-3/pull/3161#discussion_r1479988155
##########
modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItTransactionRecoveryTest.java:
##########
@@ -771,6 +784,153 @@ public void testPrimaryFailureWhileInflightInProgressAfterFirstResponse() throws
assertNull(rec);
}
+ @Test
+ public void testTsRecoveryForCursor() throws Exception {
+ TableImpl tbl = (TableImpl) node(0).tables().table(TABLE_NAME);
+
+ RecordView view1 = node(0).tables().table(TABLE_NAME).recordView();
+
+ for (int i = 0; i < 10; i++) {
+ view1.upsert(null, Tuple.create().set("key", i).set("val", "preload"));
+ }
+
+ var tblReplicationGrp = new TablePartitionId(tbl.tableId(), 0);
+
+ String leaseholder = waitAndGetLeaseholder(node(0), tblReplicationGrp);
+
+ IgniteImpl commitPartNode = commitPartitionPrimaryNode(leaseholder);
+
+ log.info("Transaction commit partition is determined [node={}].", commitPartNode.name());
+
+ IgniteImpl txCrdNode = nonPrimaryNode(leaseholder);
+
+ log.info("Transaction coordinator is chosen [node={}].", txCrdNode.name());
+
+ UUID orphanTxId = startTransactionWithCursorAndStopNode(txCrdNode);
+
+ IgniteImpl newCoordNode = node(0);
+
+ log.info("New transaction coordinator is chosen [node={}].", newCoordNode.name());
+
+ CompletableFuture<Void> txMsgCaptureFut = new CompletableFuture<>();
+
+ commitPartNode.dropMessages((nodeName, msg) -> {
+ if (msg instanceof TxRecoveryMessage) {
+ txMsgCaptureFut.complete(null);
+ }
+
+ return false;
+ });
+
+ Transaction tx = newCoordNode.transactions().begin();
+
+ RecordView view = newCoordNode.tables().table(TABLE_NAME).recordView();
+
+ var opFut = view.upsertAsync(tx, Tuple.create().set("key", 42).set("val", "new"));
+
+ try {
+ opFut.get();
+ } catch (Exception ex) {
+ log.info("Expected conflict that have to start recovery: " + ex.getMessage());
+ }
+
+ assertThat(txMsgCaptureFut, willCompleteSuccessfully());
+ }
+
+ private UUID startTransactionWithCursorAndStopNode(IgniteImpl txCrdNode) throws Exception {
+ InternalTransaction rwTx = (InternalTransaction) txCrdNode.transactions().begin();
+
+ scanSingleEntryAndLeaveCursorOpen((TableViewInternal) txCrdNode.tables().table(TABLE_NAME), rwTx, null, null);
+
+ String txCrdNodeId = txCrdNode.id();
+
+ txCrdNode.stop();
+
+ assertTrue(waitForCondition(
+ () -> node(0).clusterNodes().stream().filter(n -> txCrdNodeId.equals(n.id())).count() == 0,
+ 10_000)
+ );
+
+ return rwTx.id();
+ }
+
+ /**
+ * Starts a scan procedure for a specific transaction and reads only the first line from the cursor.
+ *
+ * @param tbl Scanned table.
+ * @param tx Transaction.
+ * @param idxId Index id.
+ * @throws Exception If failed.
+ */
+ private void scanSingleEntryAndLeaveCursorOpen(TableViewInternal tbl, InternalTransaction tx, Integer idxId, BinaryTuple exactKey)
Review Comment:
Make this method more simple because the idxId and exactKey parameters are not used.
##########
modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItTransactionRecoveryTest.java:
##########
@@ -85,6 +96,8 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
/** Table name. */
private static final String TABLE_NAME = "test_table";
+ private static final int PART_ID = 0;
Review Comment:
Please use this constant in other places where the partition ID is expected.
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/HeapUnboundedLockManager.java:
##########
@@ -500,8 +516,13 @@ public Waiter waiter(UUID txId) {
* @param acquirerTx Transaction which tries to acquire the lock.
* @param holderTx Transaction which holds the lock.
*/
- private void conflictFound(UUID acquirerTx, UUID holderTx) {
- fireEvent(LockEvent.LOCK_CONFLICT, new LockEventParameters(acquirerTx, holderTx));
+ private boolean conflictFound(UUID acquirerTx, UUID holderTx) {
+ CompletableFuture<Void> eventResult = fireEvent(LockEvent.LOCK_CONFLICT, new LockEventParameters(acquirerTx, holderTx));
+ // No async handling is expected.
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-21153
+ assert eventResult.isDone();
Review Comment:
Add a message specific to this assertion.
--
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-21248 Add abandoned locks handling to HeapUnboundedLockManager [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #3161:
URL: https://github.com/apache/ignite-3/pull/3161#discussion_r1480990730
##########
modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/HeapUnboundedLockManager.java:
##########
@@ -500,8 +516,13 @@ public Waiter waiter(UUID txId) {
* @param acquirerTx Transaction which tries to acquire the lock.
* @param holderTx Transaction which holds the lock.
*/
- private void conflictFound(UUID acquirerTx, UUID holderTx) {
- fireEvent(LockEvent.LOCK_CONFLICT, new LockEventParameters(acquirerTx, holderTx));
+ private boolean conflictFound(UUID acquirerTx, UUID holderTx) {
+ CompletableFuture<Void> eventResult = fireEvent(LockEvent.LOCK_CONFLICT, new LockEventParameters(acquirerTx, holderTx));
+ // No async handling is expected.
+ // TODO: https://issues.apache.org/jira/browse/IGNITE-21153
+ assert eventResult.isDone();
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-21248 Add abandoned locks handling to HeapUnboundedLockManager [ignite-3]
Posted by "Cyrill (via GitHub)" <gi...@apache.org>.
Cyrill commented on code in PR #3161:
URL: https://github.com/apache/ignite-3/pull/3161#discussion_r1480996918
##########
modules/table/src/integrationTest/java/org/apache/ignite/internal/table/ItTransactionRecoveryTest.java:
##########
@@ -85,6 +96,8 @@ public class ItTransactionRecoveryTest extends ClusterPerTestIntegrationTest {
/** Table name. */
private static final String TABLE_NAME = "test_table";
+ private static final int PART_ID = 0;
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