This is an automated email from the ASF dual-hosted git repository.
vpyatkov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new b129fd5adc IGNITE-20875 Add enlistment consistency token to
PrimaryReplicaReques… (#2883)
b129fd5adc is described below
commit b129fd5adca7fa97be591ba862c0aae41e54bc9c
Author: Vladislav Pyatkov <[email protected]>
AuthorDate: Wed Nov 29 14:22:43 2023 +0300
IGNITE-20875 Add enlistment consistency token to PrimaryReplicaReques…
(#2883)
---
.../ItPlacementDriverReplicaSideTest.java | 2 ++
.../replicator/message/PrimaryReplicaRequest.java | 7 ++++
.../message/ReadOnlyDirectReplicaRequest.java | 9 +----
.../sql/engine/exec/UpdatableTableImpl.java | 6 ++--
.../ignite/distributed/ReplicaUnavailableTest.java | 1 +
.../request/BuildIndexReplicaRequest.java | 7 ++--
.../request/ReadWriteReplicaRequest.java | 11 ------
.../replicator/PartitionReplicaListener.java | 41 ++++++++-------------
.../distributed/storage/InternalTableImpl.java | 28 +++++++--------
.../PartitionReplicaListenerIndexLockingTest.java | 8 ++---
.../replication/PartitionReplicaListenerTest.java | 42 +++++++++++-----------
.../ignite/internal/tx/impl/TxManagerImpl.java | 2 +-
.../tx/message/TxFinishReplicaRequest.java | 13 ++-----
.../internal/tx/message/TxRecoveryMessage.java | 11 +-----
14 files changed, 73 insertions(+), 115 deletions(-)
diff --git
a/modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java
b/modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java
index 72e0dae931..29087c5199 100644
---
a/modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java
+++
b/modules/replicator/src/integrationTest/java/org/apache/ignite/internal/replicator/ItPlacementDriverReplicaSideTest.java
@@ -276,6 +276,7 @@ public class ItPlacementDriverReplicaSideTest extends
IgniteAbstractTest {
new ReplicaService(clusterService.messagingService(), clock).invoke(
clusterService.topologyService().getByConsistentId(leaderNodeName),
TEST_REPLICA_MESSAGES_FACTORY.primaryReplicaTestRequest()
+ .enlistmentConsistencyToken(1L)
.groupId(GROUP_ID)
.build()
);
@@ -335,6 +336,7 @@ public class ItPlacementDriverReplicaSideTest extends
IgniteAbstractTest {
new ReplicaService(clusterService.messagingService(), clock).invoke(
clusterService.topologyService().getByConsistentId(leaderNodeName),
TEST_REPLICA_MESSAGES_FACTORY.primaryReplicaTestRequest()
+ .enlistmentConsistencyToken(1L)
.groupId(GROUP_ID)
.build()
);
diff --git
a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/PrimaryReplicaRequest.java
b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/PrimaryReplicaRequest.java
index 9768f0fcae..1cb918f0d2 100644
---
a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/PrimaryReplicaRequest.java
+++
b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/PrimaryReplicaRequest.java
@@ -21,4 +21,11 @@ package org.apache.ignite.internal.replicator.message;
* This is an interface for requests are designed for primary replica.
*/
public interface PrimaryReplicaRequest extends ReplicaRequest {
+ /**
+ * Gets an enlistment consistency token.
+ * The token is used to check that the lease is still actual while the
message goes to the replica.
+ *
+ * @return Enlistment consistency token.
+ */
+ Long enlistmentConsistencyToken();
}
diff --git
a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/ReadOnlyDirectReplicaRequest.java
b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/ReadOnlyDirectReplicaRequest.java
index 29ca899ee2..72c8ca0222 100644
---
a/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/ReadOnlyDirectReplicaRequest.java
+++
b/modules/replicator/src/main/java/org/apache/ignite/internal/replicator/message/ReadOnlyDirectReplicaRequest.java
@@ -23,12 +23,5 @@ package org.apache.ignite.internal.replicator.message;
*
* <p>The requests are used to implement an implicit read-only transaction for
a single partition.
*/
-public interface ReadOnlyDirectReplicaRequest extends ReplicaRequest {
- /**
- * Gets an enlistment consistency token.
- * The token is used to check that the lease is still actual while the
message goes to the replica.
- *
- * @return Enlistment consistency token.
- */
- Long enlistmentConsistencyToken();
+public interface ReadOnlyDirectReplicaRequest extends PrimaryReplicaRequest {
}
diff --git
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/UpdatableTableImpl.java
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/UpdatableTableImpl.java
index 9272cc1123..5886607b73 100644
---
a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/UpdatableTableImpl.java
+++
b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/exec/UpdatableTableImpl.java
@@ -191,7 +191,7 @@ public final class UpdatableTableImpl implements
UpdatableTable {
.schemaVersion(partToRows.getValue().get(0).schemaVersion())
.binaryTuples(binaryRowsToBuffers(partToRows.getValue()))
.transactionId(txAttributes.id())
- .term(nodeWithTerm.term())
+ .enlistmentConsistencyToken(nodeWithTerm.term())
.requestType(RequestType.RW_UPSERT_ALL)
.timestampLong(clock.nowLong())
.skipDelayedAck(true)
@@ -267,7 +267,7 @@ public final class UpdatableTableImpl implements
UpdatableTable {
.schemaVersion(rowBatch.requestedRows.get(0).schemaVersion())
.binaryTuples(binaryRowsToBuffers(rowBatch.requestedRows))
.transactionId(txAttributes.id())
- .term(nodeWithTerm.term())
+ .enlistmentConsistencyToken(nodeWithTerm.term())
.requestType(RequestType.RW_INSERT_ALL)
.timestampLong(clock.nowLong())
.skipDelayedAck(true)
@@ -336,7 +336,7 @@ public final class UpdatableTableImpl implements
UpdatableTable {
.schemaVersion(partToRows.getValue().get(0).schemaVersion())
.primaryKeys(serializePrimaryKeys(partToRows.getValue()))
.transactionId(txAttributes.id())
- .term(nodeWithTerm.term())
+ .enlistmentConsistencyToken(nodeWithTerm.term())
.requestType(RequestType.RW_DELETE_ALL)
.timestampLong(clock.nowLong())
.skipDelayedAck(true)
diff --git
a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ReplicaUnavailableTest.java
b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ReplicaUnavailableTest.java
index 91973f706f..fb426b44f1 100644
---
a/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ReplicaUnavailableTest.java
+++
b/modules/table/src/integrationTest/java/org/apache/ignite/distributed/ReplicaUnavailableTest.java
@@ -186,6 +186,7 @@ public class ReplicaUnavailableTest extends
IgniteAbstractTest {
.schemaVersion(binaryRow.schemaVersion())
.binaryTuple(binaryRow.tupleSlice())
.requestType(RequestType.RW_GET)
+ .enlistmentConsistencyToken(1L)
.build();
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replication/request/BuildIndexReplicaRequest.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replication/request/BuildIndexReplicaRequest.java
index 5563c9e064..017b11296a 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replication/request/BuildIndexReplicaRequest.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replication/request/BuildIndexReplicaRequest.java
@@ -20,7 +20,7 @@ package
org.apache.ignite.internal.table.distributed.replication.request;
import java.util.List;
import java.util.UUID;
import
org.apache.ignite.internal.replicator.exception.PrimaryReplicaMissException;
-import org.apache.ignite.internal.replicator.message.ReplicaRequest;
+import org.apache.ignite.internal.replicator.message.PrimaryReplicaRequest;
import org.apache.ignite.internal.table.distributed.TableMessageGroup;
import org.apache.ignite.network.annotations.Transferable;
@@ -30,7 +30,7 @@ import org.apache.ignite.network.annotations.Transferable;
* <p>It is possible to receive a {@link PrimaryReplicaMissException} in
response to message processing if the leaseholder changes.</p>
*/
@Transferable(TableMessageGroup.BUILD_INDEX_REPLICA_REQUEST)
-public interface BuildIndexReplicaRequest extends ReplicaRequest {
+public interface BuildIndexReplicaRequest extends PrimaryReplicaRequest {
/** Returns index ID. */
int indexId();
@@ -39,7 +39,4 @@ public interface BuildIndexReplicaRequest extends
ReplicaRequest {
/** Returns {@code true} if this batch is the last one. */
boolean finish();
-
- /** Returns an enlistment consistency token is used to check that the
lease is still actual while the message goes to the replica. */
- long enlistmentConsistencyToken();
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replication/request/ReadWriteReplicaRequest.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replication/request/ReadWriteReplicaRequest.java
index 404869b3c3..02f9ebe807 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replication/request/ReadWriteReplicaRequest.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replication/request/ReadWriteReplicaRequest.java
@@ -20,21 +20,10 @@ package
org.apache.ignite.internal.table.distributed.replication.request;
import java.util.UUID;
import org.apache.ignite.internal.replicator.message.PrimaryReplicaRequest;
import org.apache.ignite.internal.replicator.message.TimestampAware;
-import org.jetbrains.annotations.Nullable;
/**
* Read-write replica request.
*/
public interface ReadWriteReplicaRequest extends PrimaryReplicaRequest,
TimestampAware {
UUID transactionId();
-
- /**
- * Gets a raft term.
- * TODO: A temp solution until lease-based engine will be implemented
(IGNITE-17256, IGNITE-15083, IGNITE-20377)
- *
- * @return Raft term.
- */
- @Deprecated
- @Nullable
- Long term();
}
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
index 29e9483ef0..484b4ef97b 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/replicator/PartitionReplicaListener.java
@@ -91,6 +91,7 @@ import
org.apache.ignite.internal.replicator.exception.ReplicationMaxRetriesExce
import
org.apache.ignite.internal.replicator.exception.ReplicationTimeoutException;
import
org.apache.ignite.internal.replicator.exception.UnsupportedReplicaRequestException;
import org.apache.ignite.internal.replicator.listener.ReplicaListener;
+import org.apache.ignite.internal.replicator.message.PrimaryReplicaRequest;
import
org.apache.ignite.internal.replicator.message.ReadOnlyDirectReplicaRequest;
import org.apache.ignite.internal.replicator.message.ReplicaMessagesFactory;
import org.apache.ignite.internal.replicator.message.ReplicaRequest;
@@ -3357,46 +3358,32 @@ public class PartitionReplicaListener implements
ReplicaListener {
* @return Future. The result is not {@code null} only for {@link
ReadOnlyReplicaRequest}. If {@code true}, then replica is primary.
*/
private CompletableFuture<Boolean> ensureReplicaIsPrimary(ReplicaRequest
request) {
- Long expectedTerm;
-
- // TODO: IGNITE-20875 Add enlistment consistency token to
PrimaryReplicaTestRequest interface.
- if (request instanceof ReadWriteReplicaRequest) {
- expectedTerm = ((ReadWriteReplicaRequest) request).term();
-
- assert expectedTerm != null;
- } else if (request instanceof TxFinishReplicaRequest) {
- expectedTerm = ((TxFinishReplicaRequest) request).term();
-
- assert expectedTerm != null;
- } else if (request instanceof ReadOnlyDirectReplicaRequest) {
- expectedTerm = ((ReadOnlyDirectReplicaRequest)
request).enlistmentConsistencyToken();
-
- assert expectedTerm != null;
- } else if (request instanceof BuildIndexReplicaRequest) {
- expectedTerm = ((BuildIndexReplicaRequest)
request).enlistmentConsistencyToken();
- } else if (request instanceof TxRecoveryMessage) {
- expectedTerm = ((TxRecoveryMessage)
request).enlistmentConsistencyToken();
- } else {
- expectedTerm = null;
- }
-
HybridTimestamp now = hybridClock.now();
- if (expectedTerm != null) {
+ if (request instanceof PrimaryReplicaRequest) {
+ Long enlistmentConsistencyToken = ((PrimaryReplicaRequest)
request).enlistmentConsistencyToken();
+
return placementDriver.getPrimaryReplica(replicationGroupId, now)
.thenCompose(primaryReplicaMeta -> {
if (primaryReplicaMeta == null) {
- return failedFuture(new
PrimaryReplicaMissException(localNode.name(), null, expectedTerm, null, null));
+ return failedFuture(new
PrimaryReplicaMissException(
+ localNode.name(),
+ null,
+ enlistmentConsistencyToken,
+ null,
+ null
+ ));
}
long currentEnlistmentConsistencyToken =
primaryReplicaMeta.getStartTime().longValue();
// TODO:
https://issues.apache.org/jira/browse/IGNITE-20377
- if (expectedTerm != currentEnlistmentConsistencyToken
|| primaryReplicaMeta.getExpirationTime().before(now)) {
+ if (enlistmentConsistencyToken !=
currentEnlistmentConsistencyToken
+ ||
primaryReplicaMeta.getExpirationTime().before(now)) {
return failedFuture(new
PrimaryReplicaMissException(
localNode.name(),
primaryReplicaMeta.getLeaseholder(),
- expectedTerm,
+ enlistmentConsistencyToken,
currentEnlistmentConsistencyToken,
null
));
diff --git
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
index e19879c791..92531ebf44 100644
---
a/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
+++
b/modules/table/src/main/java/org/apache/ignite/internal/table/distributed/storage/InternalTableImpl.java
@@ -424,7 +424,7 @@ public class InternalTableImpl implements InternalTable {
.columnsToInclude(columnsToInclude)
.full(implicit) // Intent for one phase commit.
.batchSize(batchSize)
- .term(term)
+ .enlistmentConsistencyToken(term)
.commitPartitionId(serializeTablePartitionId(tx.commitPartition()))
.build();
@@ -747,7 +747,7 @@ public class InternalTableImpl implements InternalTable {
.primaryKey(keyRow.tupleSlice())
.commitPartitionId(serializeTablePartitionId(txo.commitPartition()))
.transactionId(txo.id())
- .term(term)
+ .enlistmentConsistencyToken(term)
.requestType(RW_GET)
.timestampLong(clock.nowLong())
.full(tx == null)
@@ -875,7 +875,7 @@ public class InternalTableImpl implements InternalTable {
.schemaVersion(rows.iterator().next().schemaVersion())
.primaryKeys(serializeBinaryTuples(rows))
.transactionId(tx.id())
- .term(term)
+ .enlistmentConsistencyToken(term)
.requestType(requestType)
.timestampLong(clock.nowLong())
.full(full)
@@ -941,7 +941,7 @@ public class InternalTableImpl implements InternalTable {
.schemaVersion(row.schemaVersion())
.binaryTuple(row.tupleSlice())
.transactionId(txo.id())
- .term(term)
+ .enlistmentConsistencyToken(term)
.requestType(RequestType.RW_UPSERT)
.timestampLong(clock.nowLong())
.full(tx == null)
@@ -992,7 +992,7 @@ public class InternalTableImpl implements InternalTable {
.schemaVersion(row.schemaVersion())
.binaryTuple(row.tupleSlice())
.transactionId(txo.id())
- .term(term)
+ .enlistmentConsistencyToken(term)
.requestType(RequestType.RW_GET_AND_UPSERT)
.timestampLong(clock.nowLong())
.full(tx == null)
@@ -1013,7 +1013,7 @@ public class InternalTableImpl implements InternalTable {
.schemaVersion(row.schemaVersion())
.binaryTuple(row.tupleSlice())
.transactionId(txo.id())
- .term(term)
+ .enlistmentConsistencyToken(term)
.requestType(RequestType.RW_INSERT)
.timestampLong(clock.nowLong())
.full(tx == null)
@@ -1061,7 +1061,7 @@ public class InternalTableImpl implements InternalTable {
.schemaVersion(rows.iterator().next().schemaVersion())
.binaryTuples(serializeBinaryTuples(rows))
.transactionId(tx.id())
- .term(term)
+ .enlistmentConsistencyToken(term)
.requestType(requestType)
.timestampLong(clock.nowLong())
.full(full)
@@ -1080,7 +1080,7 @@ public class InternalTableImpl implements InternalTable {
.schemaVersion(row.schemaVersion())
.binaryTuple(row.tupleSlice())
.transactionId(txo.id())
- .term(term)
+ .enlistmentConsistencyToken(term)
.requestType(RequestType.RW_REPLACE_IF_EXIST)
.timestampLong(clock.nowLong())
.full(tx == null)
@@ -1105,7 +1105,7 @@ public class InternalTableImpl implements InternalTable {
.oldBinaryTuple(oldRow.tupleSlice())
.newBinaryTuple(newRow.tupleSlice())
.transactionId(txo.id())
- .term(term)
+ .enlistmentConsistencyToken(term)
.requestType(RequestType.RW_REPLACE)
.timestampLong(clock.nowLong())
.full(tx == null)
@@ -1126,7 +1126,7 @@ public class InternalTableImpl implements InternalTable {
.schemaVersion(row.schemaVersion())
.binaryTuple(row.tupleSlice())
.transactionId(txo.id())
- .term(term)
+ .enlistmentConsistencyToken(term)
.requestType(RequestType.RW_GET_AND_REPLACE)
.timestampLong(clock.nowLong())
.full(tx == null)
@@ -1147,7 +1147,7 @@ public class InternalTableImpl implements InternalTable {
.schemaVersion(keyRow.schemaVersion())
.primaryKey(keyRow.tupleSlice())
.transactionId(txo.id())
- .term(term)
+ .enlistmentConsistencyToken(term)
.requestType(RequestType.RW_DELETE)
.timestampLong(clock.nowLong())
.full(tx == null)
@@ -1168,7 +1168,7 @@ public class InternalTableImpl implements InternalTable {
.schemaVersion(oldRow.schemaVersion())
.binaryTuple(oldRow.tupleSlice())
.transactionId(txo.id())
- .term(term)
+ .enlistmentConsistencyToken(term)
.requestType(RequestType.RW_DELETE_EXACT)
.timestampLong(clock.nowLong())
.full(tx == null)
@@ -1189,7 +1189,7 @@ public class InternalTableImpl implements InternalTable {
.schemaVersion(row.schemaVersion())
.primaryKey(row.tupleSlice())
.transactionId(txo.id())
- .term(term)
+ .enlistmentConsistencyToken(term)
.requestType(RequestType.RW_GET_AND_DELETE)
.timestampLong(clock.nowLong())
.full(tx == null)
@@ -1426,7 +1426,7 @@ public class InternalTableImpl implements InternalTable {
.flags(flags)
.columnsToInclude(columnsToInclude)
.batchSize(batchSize)
- .term(recipient.term())
+ .enlistmentConsistencyToken(recipient.term())
.full(false) // Set explicitly.
.commitPartitionId(serializeTablePartitionId(commitPartition))
.build();
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
index 95be4b2e27..6324fb8aa7 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerIndexLockingTest.java
@@ -297,7 +297,7 @@ public class PartitionReplicaListenerIndexLockingTest
extends IgniteAbstractTest
case RW_GET_AND_DELETE:
request =
TABLE_MESSAGES_FACTORY.readWriteSingleRowPkReplicaRequest()
.groupId(PARTITION_ID)
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(tablePartitionId(PARTITION_ID))
.transactionId(TRANSACTION_ID)
.schemaVersion(testPk.schemaVersion())
@@ -315,7 +315,7 @@ public class PartitionReplicaListenerIndexLockingTest
extends IgniteAbstractTest
case RW_GET_AND_UPSERT:
request =
TABLE_MESSAGES_FACTORY.readWriteSingleRowReplicaRequest()
.groupId(PARTITION_ID)
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(tablePartitionId(PARTITION_ID))
.transactionId(TRANSACTION_ID)
.schemaVersion(testBinaryRow.schemaVersion())
@@ -379,7 +379,7 @@ public class PartitionReplicaListenerIndexLockingTest
extends IgniteAbstractTest
case RW_DELETE_ALL:
request =
TABLE_MESSAGES_FACTORY.readWriteMultiRowPkReplicaRequest()
.groupId(PARTITION_ID)
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(tablePartitionId(PARTITION_ID))
.transactionId(TRANSACTION_ID)
.schemaVersion(pks.iterator().next().schemaVersion())
@@ -394,7 +394,7 @@ public class PartitionReplicaListenerIndexLockingTest
extends IgniteAbstractTest
case RW_UPSERT_ALL:
request =
TABLE_MESSAGES_FACTORY.readWriteMultiRowReplicaRequest()
.groupId(PARTITION_ID)
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(tablePartitionId(PARTITION_ID))
.transactionId(TRANSACTION_ID)
.schemaVersion(rows.iterator().next().schemaVersion())
diff --git
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
index 500c78ee67..32ac621d8d 100644
---
a/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
+++
b/modules/table/src/test/java/org/apache/ignite/internal/table/distributed/replication/PartitionReplicaListenerTest.java
@@ -715,7 +715,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.groupId(grpId)
.transactionId(scanTxId)
.timestampLong(clock.nowLong())
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.scanId(1L)
.indexToUse(sortedIndexId)
.batchSize(4)
@@ -732,7 +732,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.groupId(grpId)
.transactionId(scanTxId)
.timestampLong(clock.nowLong())
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.scanId(1L)
.indexToUse(sortedIndexId)
.batchSize(4)
@@ -749,7 +749,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.groupId(grpId)
.transactionId(newTxId())
.timestampLong(clock.nowLong())
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.scanId(2L)
.indexToUse(sortedIndexId)
.lowerBoundPrefix(toIndexBound(1))
@@ -769,7 +769,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.groupId(grpId)
.transactionId(newTxId())
.timestampLong(clock.nowLong())
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.scanId(2L)
.indexToUse(sortedIndexId)
.lowerBoundPrefix(toIndexBound(5))
@@ -787,7 +787,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.groupId(grpId)
.transactionId(newTxId())
.timestampLong(clock.nowLong())
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.scanId(2L)
.indexToUse(sortedIndexId)
.exactKey(toIndexKey(0))
@@ -1110,7 +1110,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.requestType(requestType)
.schemaVersion(binaryRow.schemaVersion())
.binaryTuple(binaryRow.tupleSlice())
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(commitPartitionId())
.full(full)
.build(),
@@ -1129,7 +1129,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.requestType(requestType)
.schemaVersion(binaryRow.schemaVersion())
.primaryKey(binaryRow.tupleSlice())
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(commitPartitionId())
.full(full)
.build(),
@@ -1155,7 +1155,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.requestType(requestType)
.schemaVersion(binaryRows.iterator().next().schemaVersion())
.binaryTuples(binaryRowsToBuffers(binaryRows))
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(commitPartitionId())
.full(full)
.build(),
@@ -1178,7 +1178,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.requestType(requestType)
.schemaVersion(binaryRows.iterator().next().schemaVersion())
.primaryKeys(binaryRowsToBuffers(binaryRows))
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(commitPartitionId())
.full(full)
.build(),
@@ -1202,7 +1202,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.requestType(RequestType.RW_INSERT)
.schemaVersion(binaryRow.schemaVersion())
.binaryTuple(binaryRow.tupleSlice())
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(commitPartitionId())
.build();
},
@@ -1230,7 +1230,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.requestType(RequestType.RW_UPSERT_ALL)
.schemaVersion(binaryRow0.schemaVersion())
.binaryTuples(asList(binaryRow0.tupleSlice(),
binaryRow1.tupleSlice()))
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(commitPartitionId())
.build();
},
@@ -1446,7 +1446,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.txId(txId)
.groups(Set.of(grpId))
.commit(false)
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.build();
return partitionReplicaListener.invoke(commitRequest, localNode.id());
@@ -1511,7 +1511,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.groups(Set.of(grpId))
.commit(true)
.commitTimestampLong(hybridTimestampToLong(commitTimestamp))
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.build();
return partitionReplicaListener.invoke(commitRequest, localNode.id());
@@ -1691,7 +1691,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.schemaVersion(oldRow.schemaVersion())
.oldBinaryTuple(oldRow.tupleSlice())
.newBinaryTuple(newRow.tupleSlice())
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(commitPartitionId())
.full(full)
.build(),
@@ -1708,7 +1708,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.transactionId(targetTxId)
.indexToUse(sortedIndexStorage.id())
.exactKey(toIndexKey(FUTURE_SCHEMA_ROW_INDEXED_VALUE))
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.scanId(1)
.batchSize(100)
.commitPartitionId(commitPartitionId())
@@ -1726,7 +1726,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.groupId(grpId)
.transactionId(targetTxId)
.indexToUse(sortedIndexStorage.id())
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.scanId(1)
.batchSize(100)
.commitPartitionId(commitPartitionId())
@@ -1748,7 +1748,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
TABLE_MESSAGES_FACTORY.readWriteScanRetrieveBatchReplicaRequest()
.groupId(grpId)
.transactionId(targetTxId)
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.scanId(1)
.batchSize(100)
.full(false)
@@ -1763,7 +1763,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
TABLE_MESSAGES_FACTORY.readWriteScanCloseReplicaRequest()
.groupId(grpId)
.transactionId(targetTxId)
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.scanId(1)
.build(),
localNode.id()
@@ -2230,7 +2230,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.groupId(commitPartitionId)
.groups(groups)
.txId(txId)
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commit(true)
.commitTimestampLong(clock.nowLong())
.build(),
@@ -2372,7 +2372,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.transactionId(txId)
.schemaVersion(row.schemaVersion())
.binaryTuple(row.tupleSlice())
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(commitPartitionId())
.build(),
localNode.id()
@@ -2386,7 +2386,7 @@ public class PartitionReplicaListenerTest extends
IgniteAbstractTest {
.transactionId(txId)
.schemaVersion(row.schemaVersion())
.primaryKey(row.tupleSlice())
- .term(1L)
+ .enlistmentConsistencyToken(1L)
.commitPartitionId(commitPartitionId())
.build(),
localNode.id()
diff --git
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
index fa2745e873..cacbb698c7 100644
---
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
+++
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TxManagerImpl.java
@@ -481,7 +481,7 @@ public class TxManagerImpl implements TxManager,
NetworkMessageHandler {
// In case of verification future failure transaction will be
rolled back.
.commit(commit)
.commitTimestampLong(hybridTimestampToLong(commitTimestamp))
- .term(term)
+ .enlistmentConsistencyToken(term)
.build();
return replicaService.invoke(primaryConsistentId, req)
diff --git
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxFinishReplicaRequest.java
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxFinishReplicaRequest.java
index b719eb1d95..2601f2f2f5 100644
---
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxFinishReplicaRequest.java
+++
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxFinishReplicaRequest.java
@@ -23,7 +23,7 @@ import java.util.Collection;
import java.util.UUID;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
-import org.apache.ignite.internal.replicator.message.ReplicaRequest;
+import org.apache.ignite.internal.replicator.message.PrimaryReplicaRequest;
import org.apache.ignite.internal.replicator.message.TimestampAware;
import org.apache.ignite.network.annotations.Marshallable;
import org.apache.ignite.network.annotations.Transferable;
@@ -39,7 +39,7 @@ import org.jetbrains.annotations.Nullable;
* </ol>
*/
@Transferable(value = TxMessageGroup.TX_FINISH_REQUEST)
-public interface TxFinishReplicaRequest extends ReplicaRequest, TimestampAware
{
+public interface TxFinishReplicaRequest extends PrimaryReplicaRequest,
TimestampAware {
/**
* Returns transaction Id.
*
@@ -73,13 +73,4 @@ public interface TxFinishReplicaRequest extends
ReplicaRequest, TimestampAware {
*/
@Marshallable
Collection<ReplicationGroupId> groups();
-
- /**
- * Gets a raft term.
- * TODO: A temp solution until lease-based engine will be implemented
(IGNITE-17256, IGNITE-15083)
- *
- * @return Raft term.
- */
- @Deprecated
- Long term();
}
diff --git
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxRecoveryMessage.java
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxRecoveryMessage.java
index 6f36fd8d61..a3112bba86 100644
---
a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxRecoveryMessage.java
+++
b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/message/TxRecoveryMessage.java
@@ -19,26 +19,17 @@ package org.apache.ignite.internal.tx.message;
import java.util.UUID;
import org.apache.ignite.internal.replicator.message.PrimaryReplicaRequest;
-import org.apache.ignite.internal.replicator.message.ReplicaRequest;
import org.apache.ignite.network.annotations.Transferable;
/**
* Transaction recovery message.
*/
@Transferable(TxMessageGroup.TX_RECOVERY_MSG)
-public interface TxRecoveryMessage extends PrimaryReplicaRequest,
ReplicaRequest {
+public interface TxRecoveryMessage extends PrimaryReplicaRequest {
/**
* Gets a transaction id to resolve.
*
* @return Transaction id.
*/
UUID txId();
-
- /**
- * Gets an enlistment consistency token.
- * The token is used to check that the lease is still actual while the
message goes to the replica.
- *
- * @return Enlistment consistency token.
- */
- long enlistmentConsistencyToken();
}
