You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by as...@apache.org on 2025/04/02 08:46:00 UTC
(ignite-3) branch main updated: IGNITE-24561 Use direct mapping in client transactions.
This is an automated email from the ASF dual-hosted git repository.
ascherbakov 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 e4ba67a926b IGNITE-24561 Use direct mapping in client transactions.
e4ba67a926b is described below
commit e4ba67a926b135f06ac499829a9f8ab0fbdee114
Author: Alexey Scherbakov <al...@gmail.com>
AuthorDate: Wed Apr 2 11:45:32 2025 +0300
IGNITE-24561 Use direct mapping in client transactions.
---
.../internal/client/proto/HandshakeUtils.java | 3 -
.../client/proto/ProtocolBitmaskFeature.java | 7 +-
.../internal/client/proto/tx/ClientTxUtils.java} | 14 +-
.../ignite/client/handler/ItClientHandlerTest.java | 4 +-
.../ignite/client/handler/ClientContext.java | 9 +-
.../ignite/client/handler/ClientHandlerModule.java | 3 +-
.../handler/ClientInboundMessageHandler.java | 44 ++--
.../requests/sql/ClientSqlExecuteBatchRequest.java | 2 +-
.../requests/sql/ClientSqlExecuteRequest.java | 2 +-
.../sql/ClientSqlQueryMetadataRequest.java | 2 +-
.../handler/requests/table/ClientTableCommon.java | 97 +++++---
.../table/ClientTupleContainsAllKeysRequest.java | 6 +-
.../table/ClientTupleContainsKeyRequest.java | 6 +-
.../table/ClientTupleDeleteAllExactRequest.java | 6 +-
.../table/ClientTupleDeleteAllRequest.java | 6 +-
.../table/ClientTupleDeleteExactRequest.java | 6 +-
.../requests/table/ClientTupleDeleteRequest.java | 6 +-
.../requests/table/ClientTupleGetAllRequest.java | 11 +-
.../table/ClientTupleGetAndDeleteRequest.java | 6 +-
.../table/ClientTupleGetAndReplaceRequest.java | 6 +-
.../table/ClientTupleGetAndUpsertRequest.java | 6 +-
.../requests/table/ClientTupleGetRequest.java | 7 +-
.../table/ClientTupleInsertAllRequest.java | 6 +-
.../requests/table/ClientTupleInsertRequest.java | 6 +-
.../table/ClientTupleReplaceExactRequest.java | 6 +-
.../requests/table/ClientTupleReplaceRequest.java | 6 +-
.../table/ClientTupleUpsertAllRequest.java | 6 +-
.../requests/table/ClientTupleUpsertRequest.java | 6 +-
.../requests/tx/ClientTransactionBeginRequest.java | 58 ++++-
.../tx/ClientTransactionCommitRequest.java | 111 ++++++++-
.../tx/ClientTransactionRollbackRequest.java | 33 ++-
.../ignite/internal/client/PartitionMapping.java | 60 +++++
.../ignite/internal/client/ReliableChannel.java | 78 +++++--
.../ignite/internal/client/TcpClientChannel.java | 3 +-
.../ignite/internal/client/WriteContext.java} | 15 +-
.../internal/client/compute/ClientCompute.java | 10 +-
.../client/compute/ClientJobExecution.java | 4 +
.../ignite/internal/client/sql/ClientSql.java | 10 +-
.../internal/client/table/ClientDataStreamer.java | 1 +
.../client/table/ClientKeyValueBinaryView.java | 86 +++----
.../internal/client/table/ClientKeyValueView.java | 102 ++++----
.../client/table/ClientPartitionManager.java | 8 +-
.../client/table/ClientRecordBinaryView.java | 74 +++---
.../client/table/ClientRecordSerializer.java | 18 +-
.../internal/client/table/ClientRecordView.java | 70 +++---
.../ignite/internal/client/table/ClientTable.java | 203 +++++++++++-----
.../client/table/ClientTupleSerializer.java | 116 +++++++--
.../client/table/PartitionAwarenessProvider.java | 31 +--
.../internal/client/tx/ClientLazyTransaction.java | 26 ++-
.../internal/client/tx/ClientTransaction.java | 259 +++++++++++++++++++--
.../internal/client/tx/ClientTransactions.java | 38 ++-
.../client/ObservableTimestampPropagationTest.java | 7 +-
.../ignite/client/TestClientHandlerModule.java | 6 +-
.../ignite/client/fakes/FakeInternalTable.java | 5 +
.../apache/ignite/client/fakes/FakeTxManager.java | 13 +-
.../RepeatedFinishClientTransactionTest.java | 104 ++++++++-
.../ignite/internal/jdbc/JdbcConnection.java | 2 +-
.../ReadWriteSingleRowPkReplicaRequest.java | 6 +
.../ReadWriteSingleRowReplicaRequest.java | 6 +
.../ReadWriteSwapRowReplicaRequest.java | 7 +-
.../platforms/cpp/ignite/odbc/sql_connection.cpp | 1 +
modules/runner/build.gradle | 20 +-
.../benchmark/AbstractMultiNodeBenchmark.java | 17 +-
.../internal/benchmark/ClientKvBenchmark.java | 162 +++++++++++++
.../internal/benchmark/RemoteKvBenchmark.java | 81 +++++++
.../internal/client/ItClientDirectMappingTest.java | 150 ++++++++++++
.../app/client/ItAbstractThinClientTest.java | 4 +-
.../app/client/ItThinClientTransactionsTest.java | 253 +++++++++++++++++++-
.../sql/engine/tx/QueryTransactionContextImpl.java | 2 +-
.../sql/engine/framework/NoOpTransaction.java | 5 -
.../ignite/internal/table/InternalTable.java | 8 +
.../replicator/PartitionReplicaListener.java | 58 +++--
.../distributed/storage/InternalTableImpl.java | 34 ++-
.../TransactionExpirationRegistryBenchmark.java | 5 -
.../ignite/internal/tx/InternalTransaction.java | 16 +-
.../org/apache/ignite/internal/tx/TxManager.java | 13 ++
.../tx/impl/PublicApiThreadingTransaction.java | 5 -
.../internal/tx/impl/ReadOnlyTransactionImpl.java | 6 +-
.../internal/tx/impl/ReadWriteTransactionImpl.java | 7 +-
.../tx/impl/RemoteReadWriteTransaction.java | 167 +++++++++++++
.../tx/impl/TransactionExpirationRegistry.java | 61 +++--
.../ignite/internal/tx/impl/TxManagerImpl.java | 70 ++++--
...ransactionExpirationRegistryConcurrentTest.java | 9 +-
.../tx/impl/TransactionExpirationRegistryTest.java | 15 ++
.../internal/tx/test/ItTransactionTestUtils.java | 27 +++
85 files changed, 2468 insertions(+), 593 deletions(-)
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/HandshakeUtils.java b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/HandshakeUtils.java
index 5b87b845aea..e544c4fe341 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/HandshakeUtils.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/HandshakeUtils.java
@@ -25,9 +25,6 @@ import java.util.Map;
* Utils for client protocol handshake handling.
*/
public class HandshakeUtils {
- /** Empty features. */
- public static final BitSet EMPTY_FEATURES = new BitSet();
-
/** Client type: general. */
public static final int CLIENT_TYPE_GENERAL = 2;
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ProtocolBitmaskFeature.java b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ProtocolBitmaskFeature.java
index 23ebf3620af..783c8682f19 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ProtocolBitmaskFeature.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/ProtocolBitmaskFeature.java
@@ -32,7 +32,12 @@ public enum ProtocolBitmaskFeature {
/**
* TABLE_GET/TABLES_GET requests with {@link QualifiedName}.
*/
- TABLE_GET_REQS_USE_QUALIFIED_NAME(1);
+ TABLE_GET_REQS_USE_QUALIFIED_NAME(1),
+
+ /**
+ * Direct mapping for transactions.
+ */
+ TX_DIRECT_MAPPING(2);
private static final EnumSet<ProtocolBitmaskFeature> ALL_FEATURES_AS_ENUM_SET =
EnumSet.allOf(ProtocolBitmaskFeature.class);
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSwapRowReplicaRequest.java b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/tx/ClientTxUtils.java
similarity index 60%
copy from modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSwapRowReplicaRequest.java
copy to modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/tx/ClientTxUtils.java
index 65ee84ac863..cefa3f25d14 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSwapRowReplicaRequest.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/client/proto/tx/ClientTxUtils.java
@@ -15,16 +15,12 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.partition.replicator.network.replication;
-
-import org.apache.ignite.internal.network.annotations.Transferable;
-import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
-import org.apache.ignite.internal.replicator.message.TableAware;
+package org.apache.ignite.internal.client.proto.tx;
/**
- * Read-write dual row replica request.
+ * Utility class for client transactions.
*/
-@Transferable(PartitionReplicationMessageGroup.RW_DUAL_ROW_REPLICA_REQUEST)
-public interface ReadWriteSwapRowReplicaRequest extends SwapRowReplicaRequest, ReadWriteReplicaRequest, TableAware {
-
+public class ClientTxUtils {
+ /** Tx resource id for direct mapping. */
+ public static long TX_ID_DIRECT = 0L;
}
diff --git a/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerTest.java b/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerTest.java
index bdf8e4772f1..79bdb062141 100644
--- a/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerTest.java
+++ b/modules/client-handler/src/integrationTest/java/org/apache/ignite/client/handler/ItClientHandlerTest.java
@@ -533,7 +533,7 @@ public class ItClientHandlerTest extends BaseIgniteAbstractTest {
// Server features
BitSet expected = new BitSet();
- expected.set(1);
+ expected.set(1, 3);
assertEquals(expected, supportedFeatures);
var extensionsLen = unpacker.unpackInt();
@@ -549,7 +549,7 @@ public class ItClientHandlerTest extends BaseIgniteAbstractTest {
}
}
- private void writeAndFlushLoop(Socket socket) throws Exception {
+ private static void writeAndFlushLoop(Socket socket) throws Exception {
var stop = System.currentTimeMillis() + 5000;
var out = socket.getOutputStream();
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientContext.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientContext.java
index 8b2964604d2..16eb0008ff1 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientContext.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientContext.java
@@ -18,6 +18,7 @@
package org.apache.ignite.client.handler;
import java.util.BitSet;
+import org.apache.ignite.internal.client.proto.ProtocolBitmaskFeature;
import org.apache.ignite.internal.client.proto.ProtocolVersion;
import org.apache.ignite.internal.security.authentication.UserDetails;
import org.apache.ignite.internal.tostring.S;
@@ -71,12 +72,12 @@ class ClientContext {
}
/**
- * Gets the features.
+ * Checks if a feature is enabled.
*
- * @return Features.
+ * @return {code True} if a feature is enabled.
*/
- public BitSet features() {
- return features;
+ public boolean hasFeature(ProtocolBitmaskFeature feature) {
+ return features.get(feature.featureId());
}
public UserDetails userDetails() {
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java
index 0098a641af5..8343b648233 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientHandlerModule.java
@@ -81,7 +81,8 @@ public class ClientHandlerModule implements IgniteComponent {
/** Supported server features. */
private static final BitSet SUPPORTED_FEATURES = ProtocolBitmaskFeature.featuresAsBitSet(EnumSet.of(
- ProtocolBitmaskFeature.TABLE_GET_REQS_USE_QUALIFIED_NAME
+ ProtocolBitmaskFeature.TABLE_GET_REQS_USE_QUALIFIED_NAME,
+ ProtocolBitmaskFeature.TX_DIRECT_MAPPING
));
/** Connection id generator.
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
index 6a3765bb646..6fd7a9c0088 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/ClientInboundMessageHandler.java
@@ -17,6 +17,7 @@
package org.apache.ignite.client.handler;
+import static org.apache.ignite.internal.client.proto.ProtocolBitmaskFeature.TX_DIRECT_MAPPING;
import static org.apache.ignite.internal.util.CompletableFutures.falseCompletedFuture;
import static org.apache.ignite.internal.util.IgniteUtils.firstNotNull;
import static org.apache.ignite.lang.ErrorGroups.Client.HANDSHAKE_HEADER_ERR;
@@ -695,55 +696,55 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
return ClientTableGetRequest.process(in, out, igniteTables);
case ClientOp.TUPLE_UPSERT:
- return ClientTupleUpsertRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleUpsertRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_GET:
- return ClientTupleGetRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleGetRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_UPSERT_ALL:
- return ClientTupleUpsertAllRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleUpsertAllRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_GET_ALL:
- return ClientTupleGetAllRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleGetAllRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_GET_AND_UPSERT:
- return ClientTupleGetAndUpsertRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleGetAndUpsertRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_INSERT:
- return ClientTupleInsertRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleInsertRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_INSERT_ALL:
- return ClientTupleInsertAllRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleInsertAllRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_REPLACE:
- return ClientTupleReplaceRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleReplaceRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_REPLACE_EXACT:
- return ClientTupleReplaceExactRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleReplaceExactRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_GET_AND_REPLACE:
- return ClientTupleGetAndReplaceRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleGetAndReplaceRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_DELETE:
- return ClientTupleDeleteRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleDeleteRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_DELETE_ALL:
- return ClientTupleDeleteAllRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleDeleteAllRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_DELETE_EXACT:
- return ClientTupleDeleteExactRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleDeleteExactRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_DELETE_ALL_EXACT:
- return ClientTupleDeleteAllExactRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleDeleteAllExactRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_GET_AND_DELETE:
- return ClientTupleGetAndDeleteRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleGetAndDeleteRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_CONTAINS_KEY:
- return ClientTupleContainsKeyRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleContainsKeyRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.TUPLE_CONTAINS_ALL_KEYS:
- return ClientTupleContainsAllKeysRequest.process(in, out, igniteTables, resources, txManager);
+ return ClientTupleContainsAllKeysRequest.process(in, out, igniteTables, resources, txManager, clockService);
case ClientOp.JDBC_CONNECT:
return ClientJdbcConnectRequest.execute(in, out, jdbcQueryEventHandler);
@@ -784,13 +785,16 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
return ClientJdbcPrimaryKeyMetadataRequest.process(in, out, jdbcQueryEventHandler);
case ClientOp.TX_BEGIN:
- return ClientTransactionBeginRequest.process(in, out, txManager, resources, metrics);
+ return ClientTransactionBeginRequest.process(in, out, txManager, resources, metrics, igniteTables,
+ clientContext.hasFeature(TX_DIRECT_MAPPING));
case ClientOp.TX_COMMIT:
- return ClientTransactionCommitRequest.process(in, out, resources, metrics, clockService);
+ return ClientTransactionCommitRequest.process(in, out, resources, metrics, clockService, igniteTables,
+ clientContext.hasFeature(TX_DIRECT_MAPPING));
case ClientOp.TX_ROLLBACK:
- return ClientTransactionRollbackRequest.process(in, resources, metrics);
+ return ClientTransactionRollbackRequest.process(in, resources, metrics, igniteTables,
+ clientContext.hasFeature(TX_DIRECT_MAPPING));
case ClientOp.COMPUTE_EXECUTE:
return ClientComputeExecuteRequest.process(in, out, compute, clusterService, notificationSender(requestId));
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteBatchRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteBatchRequest.java
index 7501c78be7c..13f0d4c3bf2 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteBatchRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteBatchRequest.java
@@ -50,7 +50,7 @@ public class ClientSqlExecuteBatchRequest {
QueryProcessor sql,
ClientResourceRegistry resources
) {
- InternalTransaction tx = readTx(in, out, resources);
+ InternalTransaction tx = readTx(in, out, resources, null);
ClientSqlProperties props = new ClientSqlProperties(in);
String statement = in.unpackString();
BatchedArguments arguments = in.unpackBatchedArgumentsFromBinaryTupleArray();
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteRequest.java
index 88e6155d7c8..998a31add40 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteRequest.java
@@ -76,7 +76,7 @@ public class ClientSqlExecuteRequest {
ClientResourceRegistry resources,
ClientHandlerMetricSource metrics
) {
- InternalTransaction tx = readTx(in, out, resources);
+ InternalTransaction tx = readTx(in, out, resources, null);
ClientSqlProperties props = new ClientSqlProperties(in);
String statement = in.unpackString();
Object[] arguments = in.unpackObjectArrayFromBinaryTuple();
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlQueryMetadataRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlQueryMetadataRequest.java
index 6f8509deeea..fb1e5865074 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlQueryMetadataRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlQueryMetadataRequest.java
@@ -46,7 +46,7 @@ public class ClientSqlQueryMetadataRequest {
QueryProcessor processor,
ClientResourceRegistry resources
) {
- var tx = readTx(in, out, resources);
+ var tx = readTx(in, out, resources, null);
String schema = in.unpackString();
String query = in.unpackString();
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java
index e06236a382b..74b98e2ebe6 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTableCommon.java
@@ -18,8 +18,10 @@
package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.internal.client.proto.ClientMessageCommon.NO_VALUE;
+import static org.apache.ignite.internal.client.proto.tx.ClientTxUtils.TX_ID_DIRECT;
import static org.apache.ignite.lang.ErrorGroups.Client.PROTOCOL_ERR;
import static org.apache.ignite.lang.ErrorGroups.Client.TABLE_ID_NOT_FOUND_ERR;
+import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_ALREADY_FINISHED_WITH_TIMEOUT_ERR;
import java.util.ArrayList;
import java.util.Collection;
@@ -34,10 +36,12 @@ import org.apache.ignite.internal.client.proto.ClientBinaryTupleUtils;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.hlc.HybridTimestampTracker;
import org.apache.ignite.internal.lang.IgniteInternalCheckedException;
import org.apache.ignite.internal.lang.NodeStoppingException;
+import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.schema.SchemaAware;
import org.apache.ignite.internal.schema.SchemaDescriptor;
import org.apache.ignite.internal.schema.SchemaRegistry;
@@ -45,6 +49,7 @@ import org.apache.ignite.internal.table.IgniteTablesInternal;
import org.apache.ignite.internal.table.TableViewInternal;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.InternalTxOptions;
+import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.type.DecimalNativeType;
import org.apache.ignite.internal.type.NativeType;
@@ -56,6 +61,7 @@ import org.apache.ignite.sql.ColumnType;
import org.apache.ignite.table.IgniteTables;
import org.apache.ignite.table.Tuple;
import org.apache.ignite.table.TupleHelper;
+import org.apache.ignite.tx.TransactionException;
import org.jetbrains.annotations.Nullable;
/**
@@ -65,9 +71,9 @@ public class ClientTableCommon {
/**
* Writes a schema.
*
- * @param packer Packer.
+ * @param packer Packer.
* @param schemaVer Schema version.
- * @param schema Schema.
+ * @param schema Schema.
*/
static void writeSchema(ClientMessagePacker packer, int schemaVer, SchemaDescriptor schema) {
packer.packInt(schemaVer);
@@ -99,7 +105,7 @@ public class ClientTableCommon {
* Writes a tuple.
*
* @param packer Packer.
- * @param tuple Tuple.
+ * @param tuple Tuple.
*/
public static void writeTupleOrNil(ClientMessagePacker packer, Tuple tuple, TuplePart part, SchemaRegistry schemaRegistry) {
if (tuple == null) {
@@ -115,10 +121,10 @@ public class ClientTableCommon {
/**
* Writes a tuple.
*
- * @param packer Packer.
- * @param tuple Tuple.
+ * @param packer Packer.
+ * @param tuple Tuple.
* @param skipHeader Whether to skip the tuple header.
- * @param part Which part of tuple to write.
+ * @param part Which part of tuple to write.
* @throws IgniteException on failed serialization.
*/
private static void writeTuple(
@@ -167,8 +173,8 @@ public class ClientTableCommon {
/**
* Writes multiple tuples.
*
- * @param packer Packer.
- * @param tuples Tuples.
+ * @param packer Packer.
+ * @param tuples Tuples.
* @param schemaRegistry The registry.
* @throws IgniteException on failed serialization.
*/
@@ -182,9 +188,9 @@ public class ClientTableCommon {
/**
* Writes multiple tuples.
*
- * @param packer Packer.
- * @param tuples Tuples.
- * @param part Which part of tuple to write.
+ * @param packer Packer.
+ * @param tuples Tuples.
+ * @param part Which part of tuple to write.
* @param schemaRegistry The registry.
* @throws IgniteException on failed serialization.
*/
@@ -223,9 +229,9 @@ public class ClientTableCommon {
/**
* Writes multiple tuples with null flags.
*
- * @param packer Packer.
- * @param tuples Tuples.
- * @param part Which part of tuple to write.
+ * @param packer Packer.
+ * @param tuples Tuples.
+ * @param part Which part of tuple to write.
* @param schemaRegistry The registry.
* @throws IgniteException on failed serialization.
*/
@@ -283,8 +289,8 @@ public class ClientTableCommon {
* Reads a tuple.
*
* @param unpacker Unpacker.
- * @param keyOnly Whether only key fields are expected.
- * @param schema Tuple schema.
+ * @param keyOnly Whether only key fields are expected.
+ * @param schema Tuple schema.
* @return Tuple.
*/
public static Tuple readTuple(
@@ -343,12 +349,12 @@ public class ClientTableCommon {
* Reads a table.
*
* @param unpacker Unpacker.
- * @param tables Ignite tables.
+ * @param tables Ignite tables.
* @return Table.
* @throws IgniteException If an unspecified platform exception has happened internally. Is thrown when:
- * <ul>
- * <li>the node is stopping.</li>
- * </ul>
+ * <ul>
+ * <li>the node is stopping.</li>
+ * </ul>
*/
public static CompletableFuture<TableViewInternal> readTableAsync(ClientMessageUnpacker unpacker, IgniteTables tables) {
int tableId = unpacker.unpackInt();
@@ -376,6 +382,23 @@ public class ClientTableCommon {
}
}
+ /**
+ * Writes tx metadata for a direct mapping request.
+ *
+ * @param out Packer.
+ * @param clockService Clock service.
+ * @param tx The transaction.
+ */
+ public static void writeTxMeta(ClientMessagePacker out, @Nullable ClockService clockService, InternalTransaction tx) {
+ if (tx.remote()) {
+ // Remote tx carries operation enlistment info.
+ PendingTxPartitionEnlistment token = tx.enlistedPartition(null);
+ out.packString(token.primaryNodeConsistentId());
+ out.packLong(token.consistencyToken());
+ out.meta(clockService.current());
+ }
+ }
+
/**
* Returns a new table id not found exception.
*
@@ -392,24 +415,47 @@ public class ClientTableCommon {
* @param in Unpacker.
* @param out Packer.
* @param resources Resource registry.
+ * @param txManager Tx manager.
* @return Transaction, if present, or null.
*/
public static @Nullable InternalTransaction readTx(
ClientMessageUnpacker in,
ClientMessagePacker out,
- ClientResourceRegistry resources
+ ClientResourceRegistry resources,
+ @Nullable TxManager txManager
) {
if (in.tryUnpackNil()) {
return null;
}
try {
- var tx = resources.get(in.unpackLong()).get(InternalTransaction.class);
+ long id = in.unpackLong();
+ if (id == TX_ID_DIRECT) {
+ long token = in.unpackLong();
+ UUID txId = in.unpackUuid();
+ int commitTableId = in.unpackInt();
+ int commitPart = in.unpackInt();
+ UUID coord = in.unpackUuid();
+ long timeout = in.unpackLong();
+
+ InternalTransaction remote = txManager.beginRemote(txId, new TablePartitionId(commitTableId, commitPart),
+ coord, token, timeout);
+
+ // Remote transaction will be synchronously rolled back if the timeout has exceeded.
+ if (remote.isRolledBackWithTimeoutExceeded()) {
+ throw new TransactionException(TX_ALREADY_FINISHED_WITH_TIMEOUT_ERR,
+ "Transaction is already finished [tx=" + remote + "].");
+ }
+
+ return remote;
+ }
+
+ var tx = resources.get(id).get(InternalTransaction.class);
if (tx != null && tx.isReadOnly()) {
// For read-only tx, override observable timestamp that we send to the client:
// use readTimestamp() instead of now().
- out.meta(tx.readTimestamp());
+ out.meta(tx.readTimestamp()); // TODO https://issues.apache.org/jira/browse/IGNITE-24592
}
return tx;
@@ -428,15 +474,14 @@ public class ClientTableCommon {
* @param readOnly Read only flag.
* @return Transaction.
*/
- public static @Nullable InternalTransaction readOrStartImplicitTx(
+ public static InternalTransaction readOrStartImplicitTx(
ClientMessageUnpacker in,
ClientMessagePacker out,
ClientResourceRegistry resources,
TxManager txManager,
boolean readOnly
) {
-
- InternalTransaction tx = readTx(in, out, resources);
+ InternalTransaction tx = readTx(in, out, resources, txManager);
if (tx == null) {
// Implicit transactions do not use an observation timestamp because RW never depends on it, and implicit RO is always direct.
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleContainsAllKeysRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleContainsAllKeysRequest.java
index fc8e8b54b30..adfc12eb04c 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleContainsAllKeysRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleContainsAllKeysRequest.java
@@ -20,11 +20,13 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readOrStartImplicitTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuples;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -47,7 +49,8 @@ public class ClientTupleContainsAllKeysRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
// TODO: IGNITE-23603 We have to create an implicit transaction, but leave a possibility to start RO direct.
@@ -56,6 +59,7 @@ public class ClientTupleContainsAllKeysRequest {
.recordView()
.containsAllAsync(tx, keyTuples)
.thenAccept(containsAll -> {
+ writeTxMeta(out, clockService, tx);
out.packInt(table.schemaView().lastKnownSchemaVersion());
out.packBoolean(containsAll);
})
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleContainsKeyRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleContainsKeyRequest.java
index 526a94abef4..2397fd42da1 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleContainsKeyRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleContainsKeyRequest.java
@@ -20,11 +20,13 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readOrStartImplicitTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -47,12 +49,14 @@ public class ClientTupleContainsKeyRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, true);
return readTuple(in, table, true).thenCompose(keyTuple -> {
return table.recordView().getAsync(tx, keyTuple).thenAccept(t -> {
+ writeTxMeta(out, clockService, tx);
out.packInt(table.schemaView().lastKnownSchemaVersion());
out.packBoolean(t != null);
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllExactRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllExactRequest.java
index de7d50cb1ab..8b9f459460e 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllExactRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllExactRequest.java
@@ -21,11 +21,13 @@ import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuples;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTuples;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -48,12 +50,14 @@ public class ClientTupleDeleteAllExactRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuples(in, table, false).thenCompose(tuples -> {
return table.recordView().deleteAllExactAsync(tx, tuples).thenAccept(skippedTuples -> {
+ writeTxMeta(out, clockService, tx);
writeTuples(out, skippedTuples, table.schemaView());
});
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllRequest.java
index a448f05289c..4923d0c4d9b 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteAllRequest.java
@@ -21,12 +21,14 @@ import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuples;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTuples;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -49,12 +51,14 @@ public class ClientTupleDeleteAllRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuples(in, table, true).thenCompose(tuples -> {
return table.recordView().deleteAllAsync(tx, tuples).thenAccept(skippedTuples -> {
+ writeTxMeta(out, clockService, tx);
writeTuples(out, skippedTuples, TuplePart.KEY, table.schemaView());
});
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteExactRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteExactRequest.java
index 81eaad54c69..7cd1601bdf6 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteExactRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteExactRequest.java
@@ -20,11 +20,13 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readOrStartImplicitTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -47,12 +49,14 @@ public class ClientTupleDeleteExactRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuple(in, table, false).thenCompose(tuple -> {
return table.recordView().deleteExactAsync(tx, tuple).thenAccept(res -> {
+ writeTxMeta(out, clockService, tx);
out.packInt(table.schemaView().lastKnownSchemaVersion());
out.packBoolean(res);
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteRequest.java
index d6e725728d5..d54bc0b4643 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleDeleteRequest.java
@@ -20,11 +20,13 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readOrStartImplicitTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -47,12 +49,14 @@ public class ClientTupleDeleteRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuple(in, table, true).thenCompose(tuple -> {
return table.recordView().deleteAsync(tx, tuple).thenAccept(res -> {
+ writeTxMeta(out, clockService, tx);
out.packInt(table.schemaView().lastKnownSchemaVersion());
out.packBoolean(res);
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAllRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAllRequest.java
index c299d5b23a9..76b585bd93b 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAllRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAllRequest.java
@@ -21,12 +21,14 @@ import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuples;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTuplesNullable;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -49,14 +51,17 @@ public class ClientTupleGetAllRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
// TODO: IGNITE-23603 We have to create an implicit transaction, but leave a possibility to start RO direct.
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuples(in, table, true).thenCompose(keyTuples -> {
- return table.recordView().getAllAsync(tx, keyTuples).thenAccept(tuples ->
- writeTuplesNullable(out, tuples, TuplePart.KEY_AND_VAL, table.schemaView()));
+ return table.recordView().getAllAsync(tx, keyTuples).thenAccept(tuples -> {
+ writeTxMeta(out, clockService, tx);
+ writeTuplesNullable(out, tuples, TuplePart.KEY_AND_VAL, table.schemaView());
+ });
});
});
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndDeleteRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndDeleteRequest.java
index 3f3bd11c8e0..0475597f233 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndDeleteRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndDeleteRequest.java
@@ -20,12 +20,14 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readOrStartImplicitTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -48,12 +50,14 @@ public class ClientTupleGetAndDeleteRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuple(in, table, true).thenCompose(tuple -> {
return table.recordView().getAndDeleteAsync(tx, tuple).thenAccept(resTuple -> {
+ writeTxMeta(out, clockService, tx);
ClientTableCommon.writeTupleOrNil(out, resTuple, TuplePart.KEY_AND_VAL, table.schemaView());
});
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndReplaceRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndReplaceRequest.java
index a6673e84656..d5775cf3868 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndReplaceRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndReplaceRequest.java
@@ -20,12 +20,14 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readOrStartImplicitTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -48,12 +50,14 @@ public class ClientTupleGetAndReplaceRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuple(in, table, false).thenCompose(tuple -> {
return table.recordView().getAndReplaceAsync(tx, tuple).thenAccept(resTuple -> {
+ writeTxMeta(out, clockService, tx);
ClientTableCommon.writeTupleOrNil(out, resTuple, TuplePart.KEY_AND_VAL, table.schemaView());
});
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndUpsertRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndUpsertRequest.java
index 6b83dfafc99..192562a9da1 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndUpsertRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetAndUpsertRequest.java
@@ -20,12 +20,14 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readOrStartImplicitTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -48,12 +50,14 @@ public class ClientTupleGetAndUpsertRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuple(in, table, false).thenCompose(tuple -> {
return table.recordView().getAndUpsertAsync(tx, tuple).thenAccept(resTuple -> {
+ writeTxMeta(out, clockService, tx);
ClientTableCommon.writeTupleOrNil(out, resTuple, TuplePart.KEY_AND_VAL, table.schemaView());
});
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetRequest.java
index e5b79f3c69b..54a52c75c5e 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleGetRequest.java
@@ -20,12 +20,14 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readOrStartImplicitTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -40,6 +42,7 @@ public class ClientTupleGetRequest {
* @param out Packer.
* @param tables Ignite tables.
* @param resources Resource registry.
+ * @param clockService Clock service.
* @return Future.
*/
public static CompletableFuture<Void> process(
@@ -47,12 +50,14 @@ public class ClientTupleGetRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, true);
return readTuple(in, table, true).thenCompose(keyTuple -> {
return table.recordView().getAsync(tx, keyTuple).thenAccept(t -> {
+ writeTxMeta(out, clockService, tx);
ClientTableCommon.writeTupleOrNil(out, t, TuplePart.KEY_AND_VAL, table.schemaView());
});
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertAllRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertAllRequest.java
index 32d0ca03559..a8f80a606fa 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertAllRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertAllRequest.java
@@ -21,11 +21,13 @@ import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuples;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTuples;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -48,12 +50,14 @@ public class ClientTupleInsertAllRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuples(in, table, false).thenCompose(tuples -> {
return table.recordView().insertAllAsync(tx, tuples).thenAccept(skippedTuples -> {
+ writeTxMeta(out, clockService, tx);
writeTuples(out, skippedTuples, table.schemaView());
});
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertRequest.java
index feff2bb8d25..7fc120bf9ec 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleInsertRequest.java
@@ -20,11 +20,13 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readOrStartImplicitTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -47,12 +49,14 @@ public class ClientTupleInsertRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuple(in, table, false).thenCompose(tuple -> {
return table.recordView().insertAsync(tx, tuple).thenAccept(res -> {
+ writeTxMeta(out, clockService, tx);
out.packInt(table.schemaView().lastKnownSchemaVersion());
out.packBoolean(res);
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceExactRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceExactRequest.java
index 0a100b0825e..5080b909857 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceExactRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceExactRequest.java
@@ -21,11 +21,13 @@ import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readSchema;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -48,7 +50,8 @@ public class ClientTupleReplaceExactRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
@@ -58,6 +61,7 @@ public class ClientTupleReplaceExactRequest {
var newTuple = readTuple(in, false, schema);
return table.recordView().replaceAsync(tx, oldTuple, newTuple).thenAccept(res -> {
+ writeTxMeta(out, clockService, tx);
out.packInt(table.schemaView().lastKnownSchemaVersion());
out.packBoolean(res);
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceRequest.java
index d9ea6e8e1b2..9d7182067df 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleReplaceRequest.java
@@ -20,11 +20,13 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readOrStartImplicitTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -47,12 +49,14 @@ public class ClientTupleReplaceRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuple(in, table, false).thenCompose(tuple -> {
return table.recordView().replaceAsync(tx, tuple).thenAccept(res -> {
+ writeTxMeta(out, clockService, tx);
out.packInt(table.schemaView().lastKnownSchemaVersion());
out.packBoolean(res);
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertAllRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertAllRequest.java
index a16036508e8..7f965e20e93 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertAllRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertAllRequest.java
@@ -20,11 +20,13 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readOrStartImplicitTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuples;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -47,12 +49,14 @@ public class ClientTupleUpsertAllRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuples(in, table, false).thenCompose(tuples -> {
return table.recordView().upsertAllAsync(tx, tuples).thenAccept(unused -> {
+ writeTxMeta(out, clockService, tx);
out.packInt(table.schemaView().lastKnownSchemaVersion());
});
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java
index 36341966d96..8c6eb48438c 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/table/ClientTupleUpsertRequest.java
@@ -20,11 +20,13 @@ package org.apache.ignite.client.handler.requests.table;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readOrStartImplicitTx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTableAsync;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.readTuple;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.writeTxMeta;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
+import org.apache.ignite.internal.hlc.ClockService;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.table.IgniteTables;
@@ -47,12 +49,14 @@ public class ClientTupleUpsertRequest {
ClientMessagePacker out,
IgniteTables tables,
ClientResourceRegistry resources,
- TxManager txManager
+ TxManager txManager,
+ ClockService clockService
) {
return readTableAsync(in, tables).thenCompose(table -> {
var tx = readOrStartImplicitTx(in, out, resources, txManager, false);
return readTuple(in, table, false).thenCompose(tuple -> {
return table.recordView().upsertAsync(tx, tuple).thenAccept(v -> {
+ writeTxMeta(out, clockService, tx);
out.packInt(table.schemaView().lastKnownSchemaVersion());
});
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionBeginRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionBeginRequest.java
index 6f804045b05..7776a0de657 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionBeginRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionBeginRequest.java
@@ -18,6 +18,7 @@
package org.apache.ignite.client.handler.requests.tx;
import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.startExplicitTx;
+import static org.apache.ignite.client.handler.requests.table.ClientTableCommon.tableIdNotFoundException;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientHandlerMetricSource;
@@ -27,6 +28,11 @@ import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.IgniteInternalCheckedException;
+import org.apache.ignite.internal.lang.IgniteSystemProperties;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.replicator.ZonePartitionId;
+import org.apache.ignite.internal.table.IgniteTablesInternal;
+import org.apache.ignite.internal.table.TableViewInternal;
import org.apache.ignite.internal.tx.InternalTxOptions;
import org.apache.ignite.internal.tx.TxManager;
import org.jetbrains.annotations.Nullable;
@@ -35,14 +41,18 @@ import org.jetbrains.annotations.Nullable;
* Client transaction begin request.
*/
public class ClientTransactionBeginRequest {
+ private static final boolean enabledColocation = IgniteSystemProperties.enabledColocation();
+
/**
* Processes the request.
*
- * @param in Unpacker.
- * @param out Packer.
+ * @param in Unpacker.
+ * @param out Packer.
* @param txManager Transactions.
- * @param resources Resources.
- * @param metrics Metrics.
+ * @param resources Resources.
+ * @param metrics Metrics.
+ * @param igniteTables Tables facade.
+ * @param enableDirectMapping Direct mapping feature.
* @return Future.
*/
public static @Nullable CompletableFuture<Void> process(
@@ -50,27 +60,52 @@ public class ClientTransactionBeginRequest {
ClientMessagePacker out,
TxManager txManager,
ClientResourceRegistry resources,
- ClientHandlerMetricSource metrics
+ ClientHandlerMetricSource metrics,
+ IgniteTablesInternal igniteTables,
+ boolean enableDirectMapping
) throws IgniteInternalCheckedException {
boolean readOnly = in.unpackBoolean();
long timeoutMillis = in.unpackLong();
+ long observable = in.unpackLong();
HybridTimestamp observableTs = null;
+ int tableId = -1;
+ int partition = -1;
+
if (readOnly) {
// Timestamp makes sense only for read-only transactions.
- observableTs = HybridTimestamp.nullableHybridTimestamp(in.unpackLong());
+ observableTs = HybridTimestamp.nullableHybridTimestamp(observable);
+ } else if (enableDirectMapping) {
+ // Read commit partition info.
+ // It may be not available if client has not yet loaded partition map.
+ tableId = in.unpackInt();
+ partition = in.unpackInt();
}
InternalTxOptions txOptions = InternalTxOptions.builder()
.timeoutMillis(timeoutMillis)
.build();
- // NOTE: we don't use beginAsync here because it is synchronous anyway.
var tx = startExplicitTx(out, txManager, observableTs, readOnly, txOptions);
+ // Assign commit partition at the beginning to avoid races on a client.
+ if (tableId != -1) {
+ if (enabledColocation) {
+ TableViewInternal tableView = igniteTables.cachedTable(tableId);
+
+ if (tableView == null) {
+ tx.rollback();
+ throw tableIdNotFoundException(tableId);
+ }
+
+ tx.assignCommitPartition(new ZonePartitionId(tableView.zoneId(), partition));
+ } else {
+ tx.assignCommitPartition(new TablePartitionId(tableId, partition));
+ }
+ }
+
if (readOnly) {
- // For read-only tx, override observable timestamp that we send to the client:
- // use readTimestamp() instead of now().
+ // Propagate assigned read timestamp to client to enforce serializability on subsequent reads from another node.
out.meta(tx.readTimestamp());
}
@@ -78,6 +113,11 @@ public class ClientTransactionBeginRequest {
long resourceId = resources.put(new ClientResource(tx, tx::rollbackAsync));
out.packLong(resourceId);
+ if (enableDirectMapping && !readOnly) {
+ out.packUuid(tx.id());
+ out.packUuid(tx.coordinatorId());
+ }
+
metrics.transactionsActiveIncrement();
return null;
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionCommitRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionCommitRequest.java
index 652a068e0bb..0869db225e2 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionCommitRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionCommitRequest.java
@@ -17,14 +17,28 @@
package org.apache.ignite.client.handler.requests.tx;
+import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_COMMIT_ERR;
+
+import java.util.ArrayList;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientHandlerMetricSource;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.hlc.ClockService;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.IgniteInternalCheckedException;
-import org.apache.ignite.tx.Transaction;
+import org.apache.ignite.internal.lang.IgniteTuple3;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.table.IgniteTablesInternal;
+import org.apache.ignite.internal.table.InternalTable;
+import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
+import org.apache.ignite.internal.util.ExceptionUtils;
+import org.apache.ignite.tx.TransactionException;
/**
* Client transaction commit request.
@@ -38,6 +52,8 @@ public class ClientTransactionCommitRequest {
* @param resources Resources.
* @param metrics Metrics.
* @param clockService Clock service.
+ * @param igniteTables Tables.
+ * @param enableDirectMapping Enable direct mapping flag.
* @return Future.
*/
public static CompletableFuture<Void> process(
@@ -45,11 +61,75 @@ public class ClientTransactionCommitRequest {
ClientMessagePacker out,
ClientResourceRegistry resources,
ClientHandlerMetricSource metrics,
- ClockService clockService
+ ClockService clockService,
+ IgniteTablesInternal igniteTables,
+ boolean enableDirectMapping
) throws IgniteInternalCheckedException {
long resourceId = in.unpackLong();
+ InternalTransaction tx = resources.remove(resourceId).get(InternalTransaction.class);
+
+ // Attempt to merge server and client mappings.
+ if (enableDirectMapping && !tx.isReadOnly()) {
+ long causality = in.unpackLong();
+
+ // Update causality.
+ clockService.updateClock(HybridTimestamp.hybridTimestamp(causality));
+
+ int cnt = in.unpackInt(); // Number of direct enlistments.
+
+ List<IgniteTuple3<TablePartitionId, String, Long>> list = new ArrayList<>();
+ for (int i = 0; i < cnt; i++) {
+ int tableId = in.unpackInt();
+ int partId = in.unpackInt();
+ String consistentId = in.unpackString();
+ long token = in.unpackLong();
+
+ list.add(new IgniteTuple3<>(new TablePartitionId(tableId, partId), consistentId, token));
+ }
+
+ Exception ex = null;
+
+ for (IgniteTuple3<TablePartitionId, String, Long> enlistment : list) {
+ TableViewInternal table = igniteTables.cachedTable(enlistment.get1().tableId());
+
+ if (table == null) {
+ ex = new TransactionException(TX_COMMIT_ERR, "Table not found [id=" + enlistment.get1().tableId() + ']');
+ break;
+ }
+
+ table.internalTable().targetReplicationGroupId(enlistment.get1().partitionId());
+
+ if (!merge(table.internalTable(), enlistment.get1().partitionId(), enlistment.get2(), enlistment.get3(), tx, true)) {
+ ex = new TransactionException(TX_COMMIT_ERR, "Invalid enlistment token [id=" + enlistment.get1().tableId() + ']');
+ break;
+ }
+ }
+
+ if (ex != null) {
+ // Perform enlistment for rollback.
+ for (IgniteTuple3<TablePartitionId, String, Long> enlistment : list) {
+ TableViewInternal table = igniteTables.cachedTable(enlistment.get1().tableId());
- Transaction tx = resources.remove(resourceId).get(Transaction.class);
+ if (table == null) {
+ continue;
+ }
+
+ merge(table.internalTable(), enlistment.get1().partitionId(), enlistment.get2(), enlistment.get3(), tx, false);
+ }
+
+ Exception finalEx = ex;
+
+ return tx.rollbackAsync().whenComplete((res, err) -> {
+ if (err != null) {
+ finalEx.addSuppressed(err);
+ }
+
+ metrics.transactionsActiveDecrement();
+
+ ExceptionUtils.sneakyThrow(finalEx);
+ });
+ }
+ }
return tx.commitAsync().whenComplete((res, err) -> {
if (!tx.isReadOnly()) {
@@ -59,4 +139,29 @@ public class ClientTransactionCommitRequest {
metrics.transactionsActiveDecrement();
});
}
+
+ /**
+ * Attempts to merge server and client enlistments.
+ *
+ * @param table The table.
+ * @param partId Partition id.
+ * @param consistentId Node consistent id.
+ * @param token Enlistment token.
+ * @param tx The transaction.
+ * @param commit Merge mode: commit (with validation) or rollback (no validations).
+ *
+ * @return {@code True} if merged.
+ */
+ public static boolean merge(InternalTable table, int partId, String consistentId, long token, InternalTransaction tx, boolean commit) {
+ ReplicationGroupId replicationGroupId = table.targetReplicationGroupId(partId);
+ PendingTxPartitionEnlistment existing = tx.enlistedPartition(replicationGroupId);
+ if (existing == null) {
+ tx.enlist(replicationGroupId, table.tableId(), consistentId, token);
+ } else {
+ // Enlistment tokens should be equal on commit.
+ return !commit || existing.consistencyToken() == token;
+ }
+
+ return true;
+ }
}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionRollbackRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionRollbackRequest.java
index 9f0875855c2..77c61afd35d 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionRollbackRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/tx/ClientTransactionRollbackRequest.java
@@ -17,12 +17,16 @@
package org.apache.ignite.client.handler.requests.tx;
+import static org.apache.ignite.client.handler.requests.tx.ClientTransactionCommitRequest.merge;
+
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.client.handler.ClientHandlerMetricSource;
import org.apache.ignite.client.handler.ClientResourceRegistry;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.lang.IgniteInternalCheckedException;
-import org.apache.ignite.tx.Transaction;
+import org.apache.ignite.internal.table.IgniteTablesInternal;
+import org.apache.ignite.internal.table.TableViewInternal;
+import org.apache.ignite.internal.tx.InternalTransaction;
/**
* Client transaction rollback request.
@@ -34,17 +38,38 @@ public class ClientTransactionRollbackRequest {
* @param in Unpacker.
* @param resources Resources.
* @param metrics Metrics.
+ * @param igniteTables Tables facade.
+ * @param enableDirectMapping Enable direct mapping.
* @return Future.
*/
public static CompletableFuture<Void> process(
ClientMessageUnpacker in,
ClientResourceRegistry resources,
- ClientHandlerMetricSource metrics)
+ ClientHandlerMetricSource metrics,
+ IgniteTablesInternal igniteTables,
+ boolean enableDirectMapping)
throws IgniteInternalCheckedException {
long resourceId = in.unpackLong();
- Transaction t = resources.remove(resourceId).get(Transaction.class);
+ InternalTransaction tx = resources.remove(resourceId).get(InternalTransaction.class);
+
+ if (enableDirectMapping && !tx.isReadOnly()) {
+ // Attempt to merge server and client transactions.
+ int cnt = in.unpackInt(); // Number of direct enlistments.
+ for (int i = 0; i < cnt; i++) {
+ int tableId = in.unpackInt();
+ int partId = in.unpackInt();
+ String consistentId = in.unpackString();
+ long token = in.unpackLong();
+
+ TableViewInternal table = igniteTables.cachedTable(tableId);
+
+ if (table != null) {
+ merge(table.internalTable(), partId, consistentId, token, tx, false);
+ }
+ }
+ }
- return t.rollbackAsync().whenComplete((res, err) -> metrics.transactionsActiveDecrement());
+ return tx.rollbackAsync().whenComplete((res, err) -> metrics.transactionsActiveDecrement());
}
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/PartitionMapping.java b/modules/client/src/main/java/org/apache/ignite/internal/client/PartitionMapping.java
new file mode 100644
index 00000000000..37fc367e570
--- /dev/null
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/PartitionMapping.java
@@ -0,0 +1,60 @@
+/*
+ * 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.client;
+
+import org.apache.ignite.internal.tostring.S;
+
+/**
+ * Partition mapping.
+ */
+public class PartitionMapping {
+ private final int tableId;
+ private final String nodeConsistentId;
+ private final int partition;
+
+ /**
+ * The constructor.
+ *
+ * @param tableId Table id
+ * @param nodeConsistentId Node consistent id.
+ * @param partition The partition.
+ */
+ public PartitionMapping(int tableId, String nodeConsistentId, int partition) {
+ this.tableId = tableId;
+ this.nodeConsistentId = nodeConsistentId;
+ this.partition = partition;
+ }
+
+ public int tableId() {
+ return tableId;
+ }
+
+ public String nodeConsistentId() {
+ return nodeConsistentId;
+ }
+
+ public int partition() {
+ return partition;
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public String toString() {
+ return S.toString(PartitionMapping.class, this);
+ }
+}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java b/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java
index 0d3b04669d2..78c3e401b44 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/ReliableChannel.java
@@ -51,6 +51,7 @@ import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import java.util.function.Function;
import java.util.function.ToIntFunction;
import java.util.stream.Collectors;
import org.apache.ignite.client.ClientOperationType;
@@ -206,8 +207,8 @@ public final class ReliableChannel implements AutoCloseable {
return clientCfg;
}
- public long observableTimestamp() {
- return HybridTimestamp.hybridTimestampToLong(observableTimeTracker.get());
+ public HybridTimestampTracker observableTimestamp() {
+ return observableTimeTracker;
}
/**
@@ -236,12 +237,13 @@ public final class ReliableChannel implements AutoCloseable {
/**
* Sends request and handles response asynchronously.
*
- * @param opCode Operation code.
+ * @param opCode Operation code.
* @param payloadWriter Payload writer.
* @param payloadReader Payload reader.
- * @param <T> response type.
- * @param preferredNodeName Unique name (consistent id) of the preferred target node. When a connection to the specified node exists,
- * it will be used to handle the request; otherwise, default connection will be used.
+ * @param <T> response type.
+ * @param preferredNodeName Unique name (consistent id) of the preferred target node. When a connection to the specified node
+ * exists, it will be used to handle the request; otherwise, default connection will be used.
+ * @param fallbackNodeName Fallback node name.
* @param retryPolicyOverride Retry policy override.
* @return Future for the operation.
*/
@@ -250,11 +252,12 @@ public final class ReliableChannel implements AutoCloseable {
@Nullable PayloadWriter payloadWriter,
@Nullable PayloadReader<T> payloadReader,
@Nullable String preferredNodeName,
+ @Nullable String fallbackNodeName,
@Nullable RetryPolicy retryPolicyOverride,
boolean expectNotifications
) {
return ClientFutureUtils.doWithRetryAsync(
- () -> getChannelAsync(preferredNodeName)
+ () -> getChannelAsync(preferredNodeName, fallbackNodeName)
.thenCompose(ch -> serviceAsyncInternal(opCode, payloadWriter, payloadReader, expectNotifications, ch)),
null,
ctx -> shouldRetry(opCode, ctx, retryPolicyOverride));
@@ -263,11 +266,42 @@ public final class ReliableChannel implements AutoCloseable {
/**
* Sends request and handles response asynchronously.
*
- * @param opCodeFunc Function that returns opCode.
- * @param retryOpType OpCode to use in retry.
+ * @param opCode Operation code.
* @param payloadWriter Payload writer.
* @param payloadReader Payload reader.
- * @param <T> response type.
+ * @param <T> response type.
+ * @param preferredNodeName Unique name (consistent id) of the preferred target node. When a connection to the specified node
+ * exists, it will be used to handle the request; otherwise, default connection will be used.
+ * @param fallbackNodeName Fallback node name to connect if a preferred node connection is not available.
+ * @param retryPolicyOverride Retry policy override.
+ * @return Future for the operation.
+ */
+ public <T> CompletableFuture<T> serviceAsync(
+ int opCode,
+ Function<ClientChannel, CompletableFuture<Void>> channelReadyCb,
+ @Nullable PayloadWriter payloadWriter,
+ @Nullable PayloadReader<T> payloadReader,
+ @Nullable String preferredNodeName,
+ @Nullable String fallbackNodeName,
+ @Nullable RetryPolicy retryPolicyOverride,
+ boolean expectNotifications
+ ) {
+ return ClientFutureUtils.doWithRetryAsync(
+ () -> getChannelAsync(preferredNodeName, fallbackNodeName)
+ .thenCompose(ch -> channelReadyCb.apply(ch).thenApply(ignored -> ch))
+ .thenCompose(ch -> serviceAsyncInternal(opCode, payloadWriter, payloadReader, expectNotifications, ch)),
+ null,
+ ctx -> shouldRetry(opCode, ctx, retryPolicyOverride));
+ }
+
+ /**
+ * Sends request and handles response asynchronously.
+ *
+ * @param opCodeFunc Function that returns opCode.
+ * @param retryOpType OpCode to use in retry.
+ * @param payloadWriter Payload writer.
+ * @param payloadReader Payload reader.
+ * @param <T> response type.
* @return Future for the operation.
*/
public <T> CompletableFuture<T> serviceAsync(
@@ -277,7 +311,7 @@ public final class ReliableChannel implements AutoCloseable {
@Nullable PayloadReader<T> payloadReader
) {
return ClientFutureUtils.doWithRetryAsync(
- () -> getChannelAsync(null)
+ () -> getChannelAsync(null, null)
.thenCompose(ch -> {
int opCode = opCodeFunc.applyAsInt(ch);
return serviceAsyncInternal(opCode, payloadWriter, payloadReader, false, ch);
@@ -289,10 +323,10 @@ public final class ReliableChannel implements AutoCloseable {
/**
* Sends request and handles response asynchronously.
*
- * @param opCode Operation code.
+ * @param opCode Operation code.
* @param payloadWriter Payload writer.
* @param payloadReader Payload reader.
- * @param <T> response type.
+ * @param <T> response type.
* @return Future for the operation.
*/
public <T> CompletableFuture<T> serviceAsync(
@@ -300,19 +334,19 @@ public final class ReliableChannel implements AutoCloseable {
PayloadWriter payloadWriter,
@Nullable PayloadReader<T> payloadReader
) {
- return serviceAsync(opCode, payloadWriter, payloadReader, null, null, false);
+ return serviceAsync(opCode, payloadWriter, payloadReader, null, null, null, false);
}
/**
* Sends request without payload and handles response asynchronously.
*
- * @param opCode Operation code.
+ * @param opCode Operation code.
* @param payloadReader Payload reader.
- * @param <T> Response type.
+ * @param <T> Response type.
* @return Future for the operation.
*/
public <T> CompletableFuture<T> serviceAsync(int opCode, PayloadReader<T> payloadReader) {
- return serviceAsync(opCode, null, payloadReader, null, null, false);
+ return serviceAsync(opCode, null, payloadReader, null, null, null, false);
}
private <T> CompletableFuture<T> serviceAsyncInternal(
@@ -328,11 +362,15 @@ public final class ReliableChannel implements AutoCloseable {
});
}
- private CompletableFuture<ClientChannel> getChannelAsync(@Nullable String preferredNodeName) {
+ private CompletableFuture<ClientChannel> getChannelAsync(@Nullable String preferredNodeName, @Nullable String fallbackNodeName) {
// 1. Preferred node connection.
if (preferredNodeName != null) {
ClientChannelHolder holder = nodeChannelsByName.get(preferredNodeName);
+ if (fallbackNodeName != null && holder == null) {
+ holder = nodeChannelsByName.get(fallbackNodeName);
+ }
+
if (holder != null) {
return holder.getOrCreateChannelAsync().thenCompose(ch -> {
if (ch != null) {
@@ -357,10 +395,10 @@ public final class ReliableChannel implements AutoCloseable {
/**
* Returns host:port_range address lines parsed as {@link InetSocketAddress} as a key. Value is the amount of appearences of an address
- * in {@code addrs} parameter.
+ * in {@code addrs} parameter.
*
* @return host:port_range address lines parsed as {@link InetSocketAddress} as a key. Value is the amount of appearences of an address
- * in {@code addrs} parameter.
+ * in {@code addrs} parameter.
*/
private static Map<InetSocketAddress, Integer> parsedAddresses(String[] addrs) {
if (addrs == null || addrs.length == 0) {
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java
index ca5084272cd..a9c6e2b26c6 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpClientChannel.java
@@ -85,7 +85,8 @@ class TcpClientChannel implements ClientChannel, ClientMessageHandler, ClientCon
/** Supported features. */
private static final BitSet SUPPORTED_FEATURES = ProtocolBitmaskFeature.featuresAsBitSet(EnumSet.of(
ProtocolBitmaskFeature.USER_ATTRIBUTES,
- ProtocolBitmaskFeature.TABLE_GET_REQS_USE_QUALIFIED_NAME
+ ProtocolBitmaskFeature.TABLE_GET_REQS_USE_QUALIFIED_NAME,
+ ProtocolBitmaskFeature.TX_DIRECT_MAPPING
));
/** Minimum supported heartbeat interval. */
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSwapRowReplicaRequest.java b/modules/client/src/main/java/org/apache/ignite/internal/client/WriteContext.java
similarity index 60%
copy from modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSwapRowReplicaRequest.java
copy to modules/client/src/main/java/org/apache/ignite/internal/client/WriteContext.java
index 65ee84ac863..0683a148a93 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSwapRowReplicaRequest.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/WriteContext.java
@@ -15,16 +15,15 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.partition.replicator.network.replication;
-import org.apache.ignite.internal.network.annotations.Transferable;
-import org.apache.ignite.internal.partition.replicator.network.PartitionReplicationMessageGroup;
-import org.apache.ignite.internal.replicator.message.TableAware;
+package org.apache.ignite.internal.client;
+
+import org.jetbrains.annotations.Nullable;
/**
- * Read-write dual row replica request.
+ * Write context.
*/
-@Transferable(PartitionReplicationMessageGroup.RW_DUAL_ROW_REPLICA_REQUEST)
-public interface ReadWriteSwapRowReplicaRequest extends SwapRowReplicaRequest, ReadWriteReplicaRequest, TableAware {
-
+public class WriteContext {
+ public @Nullable PartitionMapping pm;
+ public @Nullable Long enlistmentToken;
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientCompute.java b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientCompute.java
index 4d1d3f39f04..f45a69d651b 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientCompute.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientCompute.java
@@ -318,6 +318,7 @@ public class ClientCompute implements IgniteCompute {
ClientCompute::unpackSubmitTaskResult,
null,
null,
+ null,
true
);
}
@@ -334,6 +335,7 @@ public class ClientCompute implements IgniteCompute {
ClientCompute::unpackSubmitResult,
node.name(),
null,
+ null,
true
);
}
@@ -363,7 +365,7 @@ public class ClientCompute implements IgniteCompute {
return executeColocatedInternal(
t,
(outputChannel, schema) -> ClientRecordSerializer.writeRecRaw(key, keyMapper, schema, outputChannel.out(), TuplePart.KEY),
- ClientTupleSerializer.getPartitionAwarenessProvider(null, keyMapper, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keyMapper, key),
descriptor,
arg
);
@@ -378,7 +380,7 @@ public class ClientCompute implements IgniteCompute {
return executeColocatedInternal(
t,
(outputChannel, schema) -> ClientTupleSerializer.writeTupleRaw(key, schema, outputChannel, true),
- ClientTupleSerializer.getPartitionAwarenessProvider(null, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
descriptor,
arg
);
@@ -393,7 +395,7 @@ public class ClientCompute implements IgniteCompute {
) {
return t.doSchemaOutOpAsync(
ClientOp.COMPUTE_EXECUTE_COLOCATED,
- (schema, outputChannel) -> {
+ (schema, outputChannel, unused) -> {
ClientMessagePacker w = outputChannel.out();
w.packInt(t.tableId());
@@ -434,7 +436,7 @@ public class ClientCompute implements IgniteCompute {
int partitionId = ((HashPartition) partition).partitionId();
return t.doSchemaOutOpAsync(
ClientOp.COMPUTE_EXECUTE_PARTITIONED,
- (schema, outputChannel) -> {
+ (schema, outputChannel, unused) -> {
ClientMessagePacker w = outputChannel.out();
w.packInt(t.tableId());
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientJobExecution.java b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientJobExecution.java
index c78e3f31796..efcc8ba5560 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientJobExecution.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/compute/ClientJobExecution.java
@@ -121,6 +121,7 @@ class ClientJobExecution<R> implements JobExecution<R> {
ClientJobExecution::unpackJobState,
null,
null,
+ null,
false
);
}
@@ -134,6 +135,7 @@ class ClientJobExecution<R> implements JobExecution<R> {
ClientJobExecution::unpackTaskState,
null,
null,
+ null,
false
);
}
@@ -147,6 +149,7 @@ class ClientJobExecution<R> implements JobExecution<R> {
ClientJobExecution::unpackBooleanResult,
null,
null,
+ null,
false
);
}
@@ -163,6 +166,7 @@ class ClientJobExecution<R> implements JobExecution<R> {
ClientJobExecution::unpackBooleanResult,
null,
null,
+ null,
false
);
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/sql/ClientSql.java b/modules/client/src/main/java/org/apache/ignite/internal/client/sql/ClientSql.java
index a385268dcc0..b693440988e 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/sql/ClientSql.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/sql/ClientSql.java
@@ -246,7 +246,7 @@ public class ClientSql implements IgniteSql {
Objects.requireNonNull(statement);
PayloadWriter payloadWriter = w -> {
- writeTx(transaction, w);
+ writeTx(transaction, w, null);
w.out().packString(statement.defaultSchema());
w.out().packInt(statement.pageSize());
@@ -261,7 +261,7 @@ public class ClientSql implements IgniteSql {
w.out().packObjectArrayAsBinaryTuple(arguments);
- w.out().packLong(ch.observableTimestamp());
+ w.out().packLong(ch.observableTimestamp().get().longValue());
if (cancellationToken != null) {
addCancelAction(cancellationToken, w.requestId());
@@ -294,7 +294,7 @@ public class ClientSql implements IgniteSql {
@Override
public CompletableFuture<long[]> executeBatchAsync(@Nullable Transaction transaction, Statement statement, BatchedArguments batch) {
PayloadWriter payloadWriter = w -> {
- writeTx(transaction, w);
+ writeTx(transaction, w, null);
w.out().packString(statement.defaultSchema());
w.out().packInt(statement.pageSize());
@@ -306,7 +306,7 @@ public class ClientSql implements IgniteSql {
w.out().packString(statement.query());
w.out().packBatchedArgumentsAsBinaryTupleArray(batch);
- w.out().packLong(ch.observableTimestamp());
+ w.out().packLong(ch.observableTimestamp().get().longValue());
};
PayloadReader<long[]> payloadReader = r -> {
@@ -348,7 +348,7 @@ public class ClientSql implements IgniteSql {
w.out().packString(query);
w.out().packObjectArrayAsBinaryTuple(arguments);
- w.out().packLong(ch.observableTimestamp());
+ w.out().packLong(ch.observableTimestamp().get().longValue());
if (cancellationToken != null) {
addCancelAction(cancellationToken, w.requestId());
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientDataStreamer.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientDataStreamer.java
index 85cfab4dfc3..ed85f943899 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientDataStreamer.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientDataStreamer.java
@@ -97,6 +97,7 @@ class ClientDataStreamer {
? StreamerReceiverSerializer.deserializeReceiverResultsOnClient(in.in())
: null,
partitionAssignment.get(partitionId),
+ null,
new RetryLimitPolicy().retryLimit(options.retryLimit()),
false)
);
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java
index 933b7c2bdff..c6ab83630c5 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueBinaryView.java
@@ -85,10 +85,10 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET,
- (s, w) -> ser.writeTuple(tx, key, s, w, true),
+ (s, w, n) -> ser.writeTuple(tx, key, s, w, n, true),
(s, r) -> ClientTupleSerializer.readValueTuple(s, r.in()),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -109,10 +109,10 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_ALL,
- (s, w) -> ser.writeTuples(tx, keys, s, w, true),
+ (s, w, n) -> ser.writeTuples(tx, keys, s, w, n, true),
(s, r) -> ClientTupleSerializer.readKvTuplesNullable(s, r.in()),
Collections.emptyMap(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keys.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keys),
tx);
}
@@ -131,10 +131,10 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
// NullableValue.get() will never return null and there is no ambiguity between value absence and null result.
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET,
- (s, w) -> ser.writeTuple(tx, key, s, w, true),
+ (s, w, n) -> ser.writeTuple(tx, key, s, w, n, true),
(s, r) -> NullableValue.of(ClientTupleSerializer.readValueTuple(s, r.in())),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -151,10 +151,10 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET,
- (s, w) -> ser.writeTuple(tx, key, s, w, true),
+ (s, w, n) -> ser.writeTuple(tx, key, s, w, n, true),
(s, r) -> ClientTupleSerializer.readValueTuple(s, r.in()),
defaultValue,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -171,9 +171,9 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_CONTAINS_KEY,
- (s, w) -> ser.writeTuple(tx, key, s, w, true),
+ (s, w, n) -> ser.writeTuple(tx, key, s, w, n, true),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -194,9 +194,9 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_CONTAINS_ALL_KEYS,
- (s, w) -> ser.writeTuples(tx, keys, s, w, true),
+ (s, w, n) -> ser.writeTuples(tx, keys, s, w, n, true),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keys.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keys),
tx);
}
@@ -214,9 +214,9 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT,
- (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
+ (s, w, n) -> ser.writeKvTuple(tx, key, val, s, w, n, false),
r -> null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -242,9 +242,9 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT_ALL,
- (s, w) -> ser.writeKvTuples(tx, pairs.entrySet(), s, w),
+ (s, w, n) -> ser.writeKvTuples(tx, pairs.entrySet(), s, w, n),
r -> null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, pairs.keySet().iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(pairs.keySet()),
tx);
}
@@ -262,10 +262,10 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_UPSERT,
- (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
+ (s, w, n) -> ser.writeKvTuple(tx, key, val, s, w, n, false),
(s, r) -> ClientTupleSerializer.readValueTuple(s, r.in()),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -285,10 +285,10 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
// NullableValue.get() will never return null and there is no ambiguity between value absence and null result.
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_UPSERT,
- (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
+ (s, w, n) -> ser.writeKvTuple(tx, key, val, s, w, n, false),
(s, r) -> NullableValue.of(ClientTupleSerializer.readValueTuple(s, r.in())),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -306,9 +306,9 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_INSERT,
- (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
+ (s, w, n) -> ser.writeKvTuple(tx, key, val, s, w, n, false),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -331,9 +331,9 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE,
- (s, w) -> ser.writeTuple(tx, key, s, w, true),
+ (s, w, n) -> ser.writeTuple(tx, key, s, w, n, true),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -345,9 +345,9 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE_EXACT,
- (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
+ (s, w, n) -> ser.writeKvTuple(tx, key, val, s, w, n, false),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -368,10 +368,10 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_DELETE_ALL,
- (s, w) -> ser.writeTuples(tx, keys, s, w, true),
+ (s, w, n) -> ser.writeTuples(tx, keys, s, w, n, true),
(s, r) -> ClientTupleSerializer.readTuples(s, r.in(), true),
Collections.emptyList(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keys.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keys),
tx);
}
@@ -388,10 +388,10 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_DELETE,
- (s, w) -> ser.writeTuple(tx, key, s, w, true),
+ (s, w, n) -> ser.writeTuple(tx, key, s, w, n, true),
(s, r) -> ClientTupleSerializer.readValueTuple(s, r.in()),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -410,10 +410,10 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
// NullableValue.get() will never return null and there is no ambiguity between value absence and null result.
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_DELETE,
- (s, w) -> ser.writeTuple(tx, key, s, w, true),
+ (s, w, n) -> ser.writeTuple(tx, key, s, w, n, true),
(s, r) -> NullableValue.of(ClientTupleSerializer.readValueTuple(s, r.in())),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -437,9 +437,9 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE,
- (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
+ (s, w, n) -> ser.writeKvTuple(tx, key, val, s, w, n, false),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -452,12 +452,12 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE_EXACT,
- (s, w) -> {
- ser.writeKvTuple(tx, key, oldVal, s, w, false);
- ser.writeKvTuple(tx, key, newVal, s, w, true);
+ (s, w, n) -> {
+ ser.writeKvTuple(tx, key, oldVal, s, w, n, false);
+ ser.writeKvTuple(tx, key, newVal, s, w, n, true);
},
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -475,10 +475,10 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_REPLACE,
- (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
+ (s, w, n) -> ser.writeKvTuple(tx, key, val, s, w, n, false),
(s, r) -> ClientTupleSerializer.readValueTuple(s, r.in()),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -498,10 +498,10 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
// NullableValue.get() will never return null and there is no ambiguity between value absence and null result.
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_REPLACE,
- (s, w) -> ser.writeKvTuple(tx, key, val, s, w, false),
+ (s, w, n) -> ser.writeKvTuple(tx, key, val, s, w, n, false),
(s, r) -> NullableValue.of(ClientTupleSerializer.readValueTuple(s, r.in())),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -519,7 +519,7 @@ public class ClientKeyValueBinaryView extends AbstractClientView<Entry<Tuple, Tu
// The batch may go to a different node when a direct connection is not available.
StreamerBatchSender<Entry<Tuple, Tuple>, Integer, Void> batchSender = (partition, items, deleted) -> tbl.doSchemaOutOpAsync(
ClientOp.STREAMER_BATCH_SEND,
- (s, w) -> ser.writeStreamerKvTuples(partition, items, deleted, s, w),
+ (s, w, n) -> ser.writeStreamerKvTuples(partition, items, deleted, s, w),
r -> null,
PartitionAwarenessProvider.of(partition),
new RetryLimitPolicy().retryLimit(opts.retryLimit()),
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java
index d9fcac662e9..adf57de6489 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientKeyValueView.java
@@ -45,6 +45,7 @@ import org.apache.ignite.internal.binarytuple.BinaryTupleBuilder;
import org.apache.ignite.internal.binarytuple.BinaryTupleReader;
import org.apache.ignite.internal.client.PayloadInputChannel;
import org.apache.ignite.internal.client.PayloadOutputChannel;
+import org.apache.ignite.internal.client.WriteContext;
import org.apache.ignite.internal.client.proto.ClientOp;
import org.apache.ignite.internal.client.proto.TuplePart;
import org.apache.ignite.internal.client.sql.ClientSql;
@@ -114,10 +115,10 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
private CompletableFuture<V> doGet(@Nullable Transaction tx, K key, String altMethod) {
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET,
- (s, w) -> keySer.writeRec(tx, key, s, w, TuplePart.KEY),
+ (s, w, n) -> keySer.writeRec(tx, key, s, w, n, TuplePart.KEY),
(s, r) -> throwIfNull(valSer.readRec(s, r.in(), TuplePart.VAL, TuplePart.KEY_AND_VAL), altMethod),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -145,10 +146,10 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
// Null means row does not exist, NullableValue.NULL means row exists, but mapped value column is null.
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET,
- (s, w) -> keySer.writeRec(tx, key, s, w, TuplePart.KEY),
+ (s, w, n) -> keySer.writeRec(tx, key, s, w, n, TuplePart.KEY),
(s, r) -> NullableValue.of(valSer.readRec(s, r.in(), TuplePart.VAL, TuplePart.KEY_AND_VAL)),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -165,10 +166,10 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET,
- (s, w) -> keySer.writeRec(tx, key, s, w, TuplePart.KEY),
+ (s, w, n) -> keySer.writeRec(tx, key, s, w, n, TuplePart.KEY),
(s, r) -> valSer.readRec(s, r.in(), TuplePart.VAL, TuplePart.KEY_AND_VAL),
defaultValue,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -189,10 +190,10 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_ALL,
- (s, w) -> keySer.writeRecs(tx, keys, s, w, TuplePart.KEY),
+ (s, w, n) -> keySer.writeRecs(tx, keys, s, w, n, TuplePart.KEY),
this::readGetAllResponse,
Collections.emptyMap(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), keys.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), keys),
tx);
}
@@ -209,9 +210,9 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_CONTAINS_KEY,
- (s, w) -> keySer.writeRec(tx, key, s, w, TuplePart.KEY),
+ (s, w, n) -> keySer.writeRec(tx, key, s, w, n, TuplePart.KEY),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -232,9 +233,9 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_CONTAINS_ALL_KEYS,
- (s, w) -> keySer.writeRecs(tx, keys, s, w, TuplePart.KEY),
+ (s, w, n) -> keySer.writeRecs(tx, keys, s, w, n, TuplePart.KEY),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), keys.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), keys),
tx);
}
@@ -253,9 +254,9 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT,
- (s, w) -> writeKeyValue(s, w, tx, key, val),
+ (s, w, n) -> writeKeyValue(s, w, n, tx, key, val),
r -> null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -281,8 +282,8 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT_ALL,
- (s, w) -> {
- writeSchemaAndTx(s, w, tx);
+ (s, w, n) -> {
+ writeSchemaAndTx(s, w, n, tx);
w.out().packInt(pairs.size());
for (Entry<K, V> e : pairs.entrySet()) {
@@ -290,7 +291,7 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
}
},
r -> null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), pairs.keySet().iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), pairs.keySet()),
tx);
}
@@ -317,10 +318,10 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
private CompletableFuture<V> doGetAndPut(@Nullable Transaction tx, K key, @Nullable V val, String altMethod) {
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_UPSERT,
- (s, w) -> writeKeyValue(s, w, tx, key, val),
+ (s, w, n) -> writeKeyValue(s, w, n, tx, key, val),
(s, r) -> throwIfNull(valSer.readRec(s, r.in(), TuplePart.VAL, TuplePart.KEY_AND_VAL), altMethod),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -349,10 +350,10 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
private CompletableFuture<NullableValue<V>> doGetNullableAndPut(@Nullable Transaction tx, K key, @Nullable V val) {
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_UPSERT,
- (s, w) -> writeKeyValue(s, w, tx, key, val),
+ (s, w, n) -> writeKeyValue(s, w, n, tx, key, val),
(s, r) -> NullableValue.of(valSer.readRec(s, r.in(), TuplePart.VAL, TuplePart.KEY_AND_VAL)),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -371,9 +372,9 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_INSERT,
- (s, w) -> writeKeyValue(s, w, tx, key, val),
+ (s, w, n) -> writeKeyValue(s, w, n, tx, key, val),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -396,9 +397,9 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE,
- (s, w) -> keySer.writeRec(tx, key, s, w, TuplePart.KEY),
+ (s, w, n) -> keySer.writeRec(tx, key, s, w, n, TuplePart.KEY),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -411,9 +412,9 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE_EXACT,
- (s, w) -> writeKeyValue(s, w, tx, key, val),
+ (s, w, n) -> writeKeyValue(s, w, n, tx, key, val),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -434,10 +435,10 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_DELETE_ALL,
- (s, w) -> keySer.writeRecs(tx, keys, s, w, TuplePart.KEY),
+ (s, w, n) -> keySer.writeRecs(tx, keys, s, w, n, TuplePart.KEY),
(s, r) -> keySer.readRecs(s, r.in(), false, TuplePart.KEY),
Collections.emptyList(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), keys.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), keys),
tx);
}
@@ -460,10 +461,10 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
private CompletableFuture<V> doGetAndRemove(@Nullable Transaction tx, K key, String altMethod) {
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_DELETE,
- (s, w) -> keySer.writeRec(tx, key, s, w, TuplePart.KEY),
+ (s, w, n) -> keySer.writeRec(tx, key, s, w, n, TuplePart.KEY),
(s, r) -> throwIfNull(valSer.readRec(s, r.in(), TuplePart.VAL, TuplePart.KEY_AND_VAL), altMethod),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -490,10 +491,10 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
private CompletableFuture<NullableValue<V>> doGetNullableAndRemove(@Nullable Transaction tx, K key) {
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_DELETE,
- (s, w) -> keySer.writeRec(tx, key, s, w, TuplePart.KEY),
+ (s, w, n) -> keySer.writeRec(tx, key, s, w, n, TuplePart.KEY),
(s, r) -> NullableValue.of(valSer.readRec(s, r.in(), TuplePart.VAL, TuplePart.KEY_AND_VAL)),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -520,9 +521,9 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE,
- (s, w) -> writeKeyValue(s, w, tx, key, val),
+ (s, w, n) -> writeKeyValue(s, w, n, tx, key, val),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -536,13 +537,13 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE_EXACT,
- (s, w) -> {
- writeSchemaAndTx(s, w, tx);
+ (s, w, n) -> {
+ writeSchemaAndTx(s, w, n, tx);
writeKeyValueRaw(s, w, key, oldVal);
writeKeyValueRaw(s, w, key, newVal);
},
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -569,10 +570,10 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
private CompletableFuture<V> doGetAndReplace(@Nullable Transaction tx, K key, V val, String altMethod) {
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_REPLACE,
- (s, w) -> writeKeyValue(s, w, tx, key, val),
+ (s, w, n) -> writeKeyValue(s, w, n, tx, key, val),
(s, r) -> throwIfNull(valSer.readRec(s, r.in(), TuplePart.VAL, TuplePart.KEY_AND_VAL), altMethod),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
@@ -601,15 +602,22 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
private CompletableFuture<NullableValue<V>> doGetNullableAndReplace(@Nullable Transaction tx, K key, V val) {
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_REPLACE,
- (s, w) -> writeKeyValue(s, w, tx, key, val),
+ (s, w, n) -> writeKeyValue(s, w, n, tx, key, val),
(s, r) -> NullableValue.of(valSer.readRec(s, r.in(), TuplePart.VAL, TuplePart.KEY_AND_VAL)),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keySer.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keySer.mapper(), key),
tx);
}
- private void writeKeyValue(ClientSchema s, PayloadOutputChannel w, @Nullable Transaction tx, K key, @Nullable V val) {
- writeSchemaAndTx(s, w, tx);
+ private void writeKeyValue(
+ ClientSchema s,
+ PayloadOutputChannel w,
+ WriteContext ctx,
+ @Nullable Transaction tx,
+ K key,
+ @Nullable V val
+ ) {
+ writeSchemaAndTx(s, w, ctx, tx);
writeKeyValueRaw(s, w, key, val);
}
@@ -632,9 +640,9 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
w.out().packBinaryTuple(builder, noValueSet);
}
- private void writeSchemaAndTx(ClientSchema s, PayloadOutputChannel w, @Nullable Transaction tx) {
+ private void writeSchemaAndTx(ClientSchema s, PayloadOutputChannel w, WriteContext ctx, @Nullable Transaction tx) {
w.out().packInt(tbl.tableId());
- writeTx(tx, w);
+ writeTx(tx, w, ctx);
w.out().packInt(s.version());
}
@@ -671,7 +679,7 @@ public class ClientKeyValueView<K, V> extends AbstractClientView<Entry<K, V>> im
// The batch may go to a different node when a direct connection is not available.
StreamerBatchSender<Entry<K, V>, Integer, Void> batchSender = (partition, items, deleted) -> tbl.doSchemaOutOpAsync(
ClientOp.STREAMER_BATCH_SEND,
- (s, w) -> {
+ (s, w, n) -> {
w.out().packInt(tbl.tableId());
w.out().packInt(partition);
w.out().packBitSetNullable(deleted);
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientPartitionManager.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientPartitionManager.java
index 7efa4999feb..1b127dbc144 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientPartitionManager.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientPartitionManager.java
@@ -103,14 +103,14 @@ class ClientPartitionManager implements PartitionManager {
Objects.requireNonNull(key, "Key is null.");
Objects.requireNonNull(mapper, "Mapper is null.");
- return getPartition(getPartitionAwarenessProvider(null, mapper, key));
+ return getPartition(getPartitionAwarenessProvider(mapper, key));
}
@Override
public CompletableFuture<Partition> partitionAsync(Tuple key) {
Objects.requireNonNull(key, "Key is null.");
- return getPartition(getPartitionAwarenessProvider(null, key));
+ return getPartition(getPartitionAwarenessProvider(key));
}
private @Nullable ClusterNode getClusterNode(Partition partition) {
@@ -160,7 +160,9 @@ class ClientPartitionManager implements PartitionManager {
private CompletableFuture<Partition> getPartition(PartitionAwarenessProvider partitionAwarenessProvider) {
return tbl.getPartitionAssignment()
.thenCompose(partitions -> tbl.getLatestSchema().thenApply(schema -> {
- Integer hash = partitionAwarenessProvider.getObjectHashCode(schema);
+ Integer hash = partitionAwarenessProvider.getObjectHashCode(schema, true);
+
+ assert hash != null;
return new HashPartition(Math.abs(hash % partitions.size()));
}));
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordBinaryView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordBinaryView.java
index 0d7a4a365f2..81fe7d94b07 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordBinaryView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordBinaryView.java
@@ -76,10 +76,10 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET,
- (s, w) -> ser.writeTuple(tx, keyRec, s, w, true),
+ (s, w, n) -> ser.writeTuple(tx, keyRec, s, w, n, true),
(s, r) -> ClientTupleSerializer.readTuple(s, r.in(), false),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keyRec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keyRec),
tx);
}
@@ -98,10 +98,10 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_ALL,
- (s, w) -> ser.writeTuples(tx, keyRecs, s, w, true),
+ (s, w, n) -> ser.writeTuples(tx, keyRecs, s, w, n, true),
(s, r) -> ClientTupleSerializer.readTuplesNullable(s, r.in()),
Collections.emptyList(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keyRecs.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keyRecs),
tx
);
}
@@ -119,9 +119,9 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_CONTAINS_KEY,
- (s, w) -> ser.writeTuple(tx, key, s, w, true),
+ (s, w, n) -> ser.writeTuple(tx, key, s, w, n, true),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(key),
tx);
}
@@ -142,9 +142,9 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_CONTAINS_ALL_KEYS,
- (s, w) -> ser.writeTuples(tx, keys, s, w, true),
+ (s, w, n) -> ser.writeTuples(tx, keys, s, w, n, true),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keys.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keys),
tx);
}
@@ -161,9 +161,9 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT,
- (s, w) -> ser.writeTuple(tx, rec, s, w),
+ (s, w, n) -> ser.writeTuple(tx, rec, s, w, n),
r -> null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, rec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(rec),
tx);
}
@@ -184,9 +184,9 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT_ALL,
- (s, w) -> ser.writeTuples(tx, recs, s, w, false),
+ (s, w, n) -> ser.writeTuples(tx, recs, s, w, n, false),
r -> null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, recs.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(recs),
tx);
}
@@ -203,10 +203,10 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_UPSERT,
- (s, w) -> ser.writeTuple(tx, rec, s, w, false),
+ (s, w, n) -> ser.writeTuple(tx, rec, s, w, n, false),
(s, r) -> ClientTupleSerializer.readTuple(s, r.in(), false),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, rec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(rec),
tx);
}
@@ -223,9 +223,9 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_INSERT,
- (s, w) -> ser.writeTuple(tx, rec, s, w, false),
+ (s, w, n) -> ser.writeTuple(tx, rec, s, w, n, false),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, rec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(rec),
tx);
}
@@ -246,10 +246,10 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_INSERT_ALL,
- (s, w) -> ser.writeTuples(tx, recs, s, w, false),
+ (s, w, n) -> ser.writeTuples(tx, recs, s, w, n, false),
(s, r) -> ClientTupleSerializer.readTuples(s, r.in()),
Collections.emptyList(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, recs.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(recs),
tx);
}
@@ -272,9 +272,9 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE,
- (s, w) -> ser.writeTuple(tx, rec, s, w, false),
+ (s, w, n) -> ser.writeTuple(tx, rec, s, w, n, false),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, rec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(rec),
tx);
}
@@ -286,12 +286,12 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE_EXACT,
- (s, w) -> {
- ser.writeTuple(tx, oldRec, s, w, false, false);
- ser.writeTuple(tx, newRec, s, w, false, true);
+ (s, w, n) -> {
+ ser.writeTuple(tx, oldRec, s, w, n, false, false);
+ ser.writeTuple(tx, newRec, s, w, n, false, true);
},
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, oldRec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(oldRec),
tx);
}
@@ -308,10 +308,10 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_REPLACE,
- (s, w) -> ser.writeTuple(tx, rec, s, w, false),
+ (s, w, n) -> ser.writeTuple(tx, rec, s, w, n, false),
(s, r) -> ClientTupleSerializer.readTuple(s, r.in(), false),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, rec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(rec),
tx);
}
@@ -328,9 +328,9 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE,
- (s, w) -> ser.writeTuple(tx, keyRec, s, w, true),
+ (s, w, n) -> ser.writeTuple(tx, keyRec, s, w, n, true),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keyRec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keyRec),
tx);
}
@@ -347,9 +347,9 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE_EXACT,
- (s, w) -> ser.writeTuple(tx, rec, s, w, false),
+ (s, w, n) -> ser.writeTuple(tx, rec, s, w, n, false),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, rec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(rec),
tx);
}
@@ -366,10 +366,10 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_DELETE,
- (s, w) -> ser.writeTuple(tx, keyRec, s, w, true),
+ (s, w, n) -> ser.writeTuple(tx, keyRec, s, w, n, true),
(s, r) -> ClientTupleSerializer.readTuple(s, r.in(), false),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keyRec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keyRec),
tx);
}
@@ -390,10 +390,10 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_DELETE_ALL,
- (s, w) -> ser.writeTuples(tx, keyRecs, s, w, true),
+ (s, w, n) -> ser.writeTuples(tx, keyRecs, s, w, n, true),
(s, r) -> ClientTupleSerializer.readTuples(s, r.in(), true),
Collections.emptyList(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, keyRecs.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(keyRecs),
tx);
}
@@ -414,10 +414,10 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_DELETE_ALL_EXACT,
- (s, w) -> ser.writeTuples(tx, recs, s, w, false),
+ (s, w, n) -> ser.writeTuples(tx, recs, s, w, n, false),
(s, r) -> ClientTupleSerializer.readTuples(s, r.in()),
Collections.emptyList(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, recs.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(recs),
tx);
}
@@ -434,7 +434,7 @@ public class ClientRecordBinaryView extends AbstractClientView<Tuple> implements
// The batch may go to a different node when a direct connection is not available.
StreamerBatchSender<Tuple, Integer, Void> batchSender = (partitionId, items, deleted) -> tbl.doSchemaOutOpAsync(
ClientOp.STREAMER_BATCH_SEND,
- (s, w) -> ser.writeStreamerTuples(partitionId, items, deleted, s, w),
+ (s, w, n) -> ser.writeStreamerTuples(partitionId, items, deleted, s, w),
r -> null,
PartitionAwarenessProvider.of(partitionId),
new RetryLimitPolicy().retryLimit(opts.retryLimit()),
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java
index e7e4ce3d43e..f883ed6b054 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordSerializer.java
@@ -27,6 +27,7 @@ import java.util.List;
import org.apache.ignite.internal.binarytuple.BinaryTupleBuilder;
import org.apache.ignite.internal.binarytuple.BinaryTupleReader;
import org.apache.ignite.internal.client.PayloadOutputChannel;
+import org.apache.ignite.internal.client.WriteContext;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
@@ -112,9 +113,16 @@ public class ClientRecordSerializer<R> {
writeRecRaw(rec, mapper, schema, out, part);
}
- void writeRec(@Nullable Transaction tx, @Nullable R rec, ClientSchema schema, PayloadOutputChannel out, TuplePart part) {
+ void writeRec(
+ @Nullable Transaction tx,
+ @Nullable R rec,
+ ClientSchema schema,
+ PayloadOutputChannel out,
+ WriteContext ctx,
+ TuplePart part
+ ) {
out.out().packInt(tableId);
- writeTx(tx, out);
+ writeTx(tx, out, ctx);
out.out().packInt(schema.version());
writeRecRaw(rec, schema, out.out(), part);
@@ -126,10 +134,11 @@ public class ClientRecordSerializer<R> {
@Nullable R rec2,
ClientSchema schema,
PayloadOutputChannel out,
+ WriteContext ctx,
TuplePart part
) {
out.out().packInt(tableId);
- writeTx(tx, out);
+ writeTx(tx, out, ctx);
out.out().packInt(schema.version());
Marshaller marshaller = schema.getMarshaller(mapper, part);
@@ -144,10 +153,11 @@ public class ClientRecordSerializer<R> {
Collection<R> recs,
ClientSchema schema,
PayloadOutputChannel out,
+ WriteContext ctx,
TuplePart part
) {
out.out().packInt(tableId);
- writeTx(tx, out);
+ writeTx(tx, out, ctx);
out.out().packInt(schema.version());
out.out().packInt(recs.size());
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java
index 6e20ec5621c..1f40e73f365 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientRecordView.java
@@ -83,10 +83,10 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET,
- (s, w) -> ser.writeRec(tx, keyRec, s, w, TuplePart.KEY),
+ (s, w, n) -> ser.writeRec(tx, keyRec, s, w, n, TuplePart.KEY),
(s, r) -> ser.readValRec(keyRec, s, r.in()),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), keyRec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), keyRec),
tx);
}
@@ -105,10 +105,10 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_ALL,
- (s, w) -> ser.writeRecs(tx, keyRecs, s, w, TuplePart.KEY),
+ (s, w, n) -> ser.writeRecs(tx, keyRecs, s, w, n, TuplePart.KEY),
(s, r) -> ser.readRecs(s, r.in(), true, TuplePart.KEY_AND_VAL),
Collections.emptyList(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), keyRecs.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), keyRecs),
tx
);
}
@@ -126,9 +126,9 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_CONTAINS_KEY,
- (s, w) -> ser.writeRec(tx, key, s, w, TuplePart.KEY),
+ (s, w, n) -> ser.writeRec(tx, key, s, w, n, TuplePart.KEY),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), key),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), key),
tx);
}
@@ -149,9 +149,9 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_CONTAINS_ALL_KEYS,
- (s, w) -> ser.writeRecs(tx, keys, s, w, TuplePart.KEY),
+ (s, w, n) -> ser.writeRecs(tx, keys, s, w, n, TuplePart.KEY),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), keys.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), keys),
tx);
}
@@ -168,9 +168,9 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT,
- (s, w) -> ser.writeRec(tx, rec, s, w, TuplePart.KEY_AND_VAL),
+ (s, w, n) -> ser.writeRec(tx, rec, s, w, n, TuplePart.KEY_AND_VAL),
r -> null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), rec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), rec),
tx);
}
@@ -191,9 +191,9 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_UPSERT_ALL,
- (s, w) -> ser.writeRecs(tx, recs, s, w, TuplePart.KEY_AND_VAL),
+ (s, w, n) -> ser.writeRecs(tx, recs, s, w, n, TuplePart.KEY_AND_VAL),
r -> null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), recs.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), recs),
tx);
}
@@ -210,10 +210,10 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_UPSERT,
- (s, w) -> ser.writeRec(tx, rec, s, w, TuplePart.KEY_AND_VAL),
+ (s, w, n) -> ser.writeRec(tx, rec, s, w, n, TuplePart.KEY_AND_VAL),
(s, r) -> ser.readValRec(rec, s, r.in()),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), rec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), rec),
tx);
}
@@ -230,9 +230,9 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_INSERT,
- (s, w) -> ser.writeRec(tx, rec, s, w, TuplePart.KEY_AND_VAL),
+ (s, w, n) -> ser.writeRec(tx, rec, s, w, n, TuplePart.KEY_AND_VAL),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), rec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), rec),
tx);
}
@@ -253,10 +253,10 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_INSERT_ALL,
- (s, w) -> ser.writeRecs(tx, recs, s, w, TuplePart.KEY_AND_VAL),
+ (s, w, n) -> ser.writeRecs(tx, recs, s, w, n, TuplePart.KEY_AND_VAL),
(s, r) -> ser.readRecs(s, r.in(), false, TuplePart.KEY_AND_VAL),
Collections.emptyList(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), recs.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), recs),
tx);
}
@@ -279,9 +279,9 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE,
- (s, w) -> ser.writeRec(tx, rec, s, w, TuplePart.KEY_AND_VAL),
+ (s, w, n) -> ser.writeRec(tx, rec, s, w, n, TuplePart.KEY_AND_VAL),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), rec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), rec),
tx);
}
@@ -293,9 +293,9 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_REPLACE_EXACT,
- (s, w) -> ser.writeRecs(tx, oldRec, newRec, s, w, TuplePart.KEY_AND_VAL),
+ (s, w, n) -> ser.writeRecs(tx, oldRec, newRec, s, w, n, TuplePart.KEY_AND_VAL),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), oldRec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), oldRec),
tx);
}
@@ -312,10 +312,10 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_REPLACE,
- (s, w) -> ser.writeRec(tx, rec, s, w, TuplePart.KEY_AND_VAL),
+ (s, w, n) -> ser.writeRec(tx, rec, s, w, n, TuplePart.KEY_AND_VAL),
(s, r) -> ser.readValRec(rec, s, r.in()),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), rec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), rec),
tx);
}
@@ -332,9 +332,9 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE,
- (s, w) -> ser.writeRec(tx, keyRec, s, w, TuplePart.KEY),
+ (s, w, n) -> ser.writeRec(tx, keyRec, s, w, n, TuplePart.KEY),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), keyRec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), keyRec),
tx);
}
@@ -351,9 +351,9 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutOpAsync(
ClientOp.TUPLE_DELETE_EXACT,
- (s, w) -> ser.writeRec(tx, rec, s, w, TuplePart.KEY_AND_VAL),
+ (s, w, n) -> ser.writeRec(tx, rec, s, w, n, TuplePart.KEY_AND_VAL),
r -> r.in().unpackBoolean(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), rec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), rec),
tx);
}
@@ -370,10 +370,10 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_GET_AND_DELETE,
- (s, w) -> ser.writeRec(tx, keyRec, s, w, TuplePart.KEY),
+ (s, w, n) -> ser.writeRec(tx, keyRec, s, w, n, TuplePart.KEY),
(s, r) -> ser.readValRec(keyRec, s, r.in()),
null,
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), keyRec),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), keyRec),
tx);
}
@@ -394,10 +394,10 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_DELETE_ALL,
- (s, w) -> ser.writeRecs(tx, keyRecs, s, w, TuplePart.KEY),
+ (s, w, n) -> ser.writeRecs(tx, keyRecs, s, w, n, TuplePart.KEY),
(s, r) -> ser.readRecs(s, r.in(), false, TuplePart.KEY),
Collections.emptyList(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), keyRecs.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), keyRecs),
tx);
}
@@ -418,10 +418,10 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
return tbl.doSchemaOutInOpAsync(
ClientOp.TUPLE_DELETE_ALL_EXACT,
- (s, w) -> ser.writeRecs(tx, recs, s, w, TuplePart.KEY_AND_VAL),
+ (s, w, n) -> ser.writeRecs(tx, recs, s, w, n, TuplePart.KEY_AND_VAL),
(s, r) -> ser.readRecs(s, r.in(), false, TuplePart.KEY_AND_VAL),
Collections.emptyList(),
- ClientTupleSerializer.getPartitionAwarenessProvider(tx, ser.mapper(), recs.iterator().next()),
+ ClientTupleSerializer.getPartitionAwarenessProvider(ser.mapper(), recs),
tx);
}
@@ -437,7 +437,7 @@ public class ClientRecordView<R> extends AbstractClientView<R> implements Record
// The batch may go to a different node when a direct connection is not available.
StreamerBatchSender<R, Integer, Void> batchSender = (partition, items, deleted) -> tbl.doSchemaOutOpAsync(
ClientOp.STREAMER_BATCH_SEND,
- (s, w) -> ser.writeStreamerRecs(partition, items, deleted, s, w),
+ (s, w, n) -> ser.writeStreamerRecs(partition, items, deleted, s, w),
r -> null,
PartitionAwarenessProvider.of(partition),
new RetryLimitPolicy().retryLimit(opts.retryLimit()),
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
index 97a4a2de5db..2ba28eca404 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTable.java
@@ -17,9 +17,18 @@
package org.apache.ignite.internal.client.table;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.function.Function.identity;
+import static org.apache.ignite.internal.client.proto.ProtocolBitmaskFeature.TX_DIRECT_MAPPING;
+import static org.apache.ignite.internal.client.proto.tx.ClientTxUtils.TX_ID_DIRECT;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.apache.ignite.internal.util.ExceptionUtils.matchAny;
+import static org.apache.ignite.internal.util.ExceptionUtils.sneakyThrow;
+import static org.apache.ignite.internal.util.ExceptionUtils.unwrapCause;
import static org.apache.ignite.lang.ErrorGroups.Client.CONNECTION_ERR;
import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
+import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_ALREADY_FINISHED_ERR;
+import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_ALREADY_FINISHED_WITH_TIMEOUT_ERR;
import java.util.ArrayList;
import java.util.List;
@@ -27,15 +36,16 @@ import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ConcurrentHashMap;
-import java.util.function.BiConsumer;
import java.util.function.BiFunction;
import java.util.function.Function;
import org.apache.ignite.client.RetryPolicy;
import org.apache.ignite.internal.client.ClientSchemaVersionMismatchException;
import org.apache.ignite.internal.client.ClientUtils;
+import org.apache.ignite.internal.client.PartitionMapping;
import org.apache.ignite.internal.client.PayloadInputChannel;
import org.apache.ignite.internal.client.PayloadOutputChannel;
import org.apache.ignite.internal.client.ReliableChannel;
+import org.apache.ignite.internal.client.WriteContext;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.ClientOp;
import org.apache.ignite.internal.client.proto.ColumnTypeConverter;
@@ -46,6 +56,7 @@ import org.apache.ignite.internal.client.tx.ClientLazyTransaction;
import org.apache.ignite.internal.client.tx.ClientTransaction;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.lang.IgniteBiTuple;
+import org.apache.ignite.internal.lang.IgniteTriConsumer;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.marshaller.MarshallersProvider;
import org.apache.ignite.internal.marshaller.UnmappedColumnsException;
@@ -268,7 +279,7 @@ public class ClientTable implements Table {
var schema = new ClientSchema(schemaVer, columns, marshallers);
if (schemaVer != targetVer) {
- schemas.put(schemaVer, CompletableFuture.completedFuture(schema));
+ schemas.put(schemaVer, completedFuture(schema));
}
synchronized (latestSchemaLock) {
@@ -291,20 +302,31 @@ public class ClientTable implements Table {
*
* @param tx Transaction.
* @param out Packer.
+ * @param ctx Write context.
*/
- public static void writeTx(@Nullable Transaction tx, PayloadOutputChannel out) {
+ public static void writeTx(@Nullable Transaction tx, PayloadOutputChannel out, @Nullable WriteContext ctx) {
if (tx == null) {
out.out().packNil();
} else {
- ClientTransaction clientTx = ClientTransaction.get(tx);
+ ClientTransaction tx0 = ClientTransaction.get(tx);
+
+ if (ctx != null && ctx.enlistmentToken != null) {
+ out.out().packLong(TX_ID_DIRECT); // For direct enlistment, pass 0 for resourceId to distinguish with proxy mode.
+ out.out().packLong(ctx.enlistmentToken);
+ out.out().packUuid(tx0.txId());
+ out.out().packInt(tx0.commitTableId());
+ out.out().packInt(tx0.commitPartition());
+ out.out().packUuid(tx0.coordinatorId());
+ out.out().packLong(tx0.timeout());
+ } else {
+ //noinspection resource
+ if (tx0.channel() != out.clientChannel()) {
+ // Do not throw IgniteClientConnectionException to avoid retry kicking in.
+ throw new IgniteException(CONNECTION_ERR, "Transaction context has been lost due to connection errors.");
+ }
- //noinspection resource
- if (clientTx.channel() != out.clientChannel()) {
- // Do not throw IgniteClientConnectionException to avoid retry kicking in.
- throw new IgniteException(CONNECTION_ERR, "Transaction context has been lost due to connection errors.");
+ out.out().packLong(tx0.id());
}
-
- out.out().packLong(clientTx.id());
}
}
@@ -321,9 +343,9 @@ public class ClientTable implements Table {
*/
public <T> CompletableFuture<T> doSchemaOutOpAsync(
int opCode,
- BiConsumer<ClientSchema, PayloadOutputChannel> writer,
+ IgniteTriConsumer<ClientSchema, PayloadOutputChannel, WriteContext> writer,
Function<PayloadInputChannel, T> reader,
- @Nullable PartitionAwarenessProvider provider,
+ PartitionAwarenessProvider provider,
@Nullable Transaction tx) {
return doSchemaOutInOpAsync(
opCode,
@@ -352,9 +374,9 @@ public class ClientTable implements Table {
*/
public <T> CompletableFuture<T> doSchemaOutOpAsync(
int opCode,
- BiConsumer<ClientSchema, PayloadOutputChannel> writer,
+ IgniteTriConsumer<ClientSchema, PayloadOutputChannel, WriteContext> writer,
Function<PayloadInputChannel, T> reader,
- @Nullable PartitionAwarenessProvider provider,
+ PartitionAwarenessProvider provider,
boolean expectNotifications,
@Nullable Transaction tx) {
return doSchemaOutInOpAsync(
@@ -384,9 +406,9 @@ public class ClientTable implements Table {
*/
<T> CompletableFuture<T> doSchemaOutOpAsync(
int opCode,
- BiConsumer<ClientSchema, PayloadOutputChannel> writer,
+ IgniteTriConsumer<ClientSchema, PayloadOutputChannel, WriteContext> writer,
Function<PayloadInputChannel, T> reader,
- @Nullable PartitionAwarenessProvider provider,
+ PartitionAwarenessProvider provider,
@Nullable RetryPolicy retryPolicyOverride,
@Nullable Transaction tx) {
return doSchemaOutInOpAsync(
@@ -416,10 +438,10 @@ public class ClientTable implements Table {
*/
<T> CompletableFuture<T> doSchemaOutInOpAsync(
int opCode,
- BiConsumer<ClientSchema, PayloadOutputChannel> writer,
+ IgniteTriConsumer<ClientSchema, PayloadOutputChannel, WriteContext> writer,
BiFunction<ClientSchema, PayloadInputChannel, T> reader,
@Nullable T defaultValue,
- @Nullable PartitionAwarenessProvider provider,
+ PartitionAwarenessProvider provider,
@Nullable Transaction tx
) {
return doSchemaOutInOpAsync(opCode, writer, reader, defaultValue, true, provider, null, null, false, tx);
@@ -443,46 +465,72 @@ public class ClientTable implements Table {
*/
private <T> CompletableFuture<T> doSchemaOutInOpAsync(
int opCode,
- BiConsumer<ClientSchema, PayloadOutputChannel> writer,
+ IgniteTriConsumer<ClientSchema, PayloadOutputChannel, WriteContext> writer,
BiFunction<ClientSchema, PayloadInputChannel, T> reader,
@Nullable T defaultValue,
boolean responseSchemaRequired,
- @Nullable PartitionAwarenessProvider provider,
+ PartitionAwarenessProvider provider,
@Nullable RetryPolicy retryPolicyOverride,
@Nullable Integer schemaVersionOverride,
boolean expectNotifications,
- @Nullable Transaction tx) {
+ @Nullable Transaction tx
+ ) {
CompletableFuture<T> fut = new CompletableFuture<>();
CompletableFuture<ClientSchema> schemaFut = getSchema(schemaVersionOverride == null ? latestSchemaVer : schemaVersionOverride);
- CompletableFuture<List<String>> partitionsFut = provider == null || !provider.isPartitionAwarenessEnabled()
- ? nullCompletedFuture()
- : getPartitionAssignment();
+ CompletableFuture<List<String>> partitionsFut = getPartitionAssignment();
// Wait for schema and partition assignment.
CompletableFuture.allOf(schemaFut, partitionsFut)
.thenCompose(v -> {
ClientSchema schema = schemaFut.getNow(null);
- String txPreferredNodeName = getPreferredNodeName(provider, partitionsFut.getNow(null), schema);
- return ClientLazyTransaction.ensureStarted(tx, ch, txPreferredNodeName).thenCompose(unused -> {
- // Update preferred node name after starting the transaction.
- // All operations for a given explicit transaction should go to the same node (tx coordinator).
- String opPreferredNodeName = getPreferredNodeName(provider, partitionsFut.getNow(null), schema);
+ PartitionMapping forCrd = getPreferredNodeName(tableId(), provider, partitionsFut.getNow(null), schema, true);
+
+ return ClientLazyTransaction.ensureStarted(tx, ch, forCrd).thenCompose(tx0 -> {
+ @Nullable PartitionMapping forOp = getPreferredNodeName(tableId(), provider, partitionsFut.getNow(null), schema,
+ tx0 == null); // Force coordinator mode for implicit transactions.
- return ch.serviceAsync(opCode,
- w -> writer.accept(schema, w),
- r -> readSchemaAndReadData(schema, r, reader, defaultValue, responseSchemaRequired),
- opPreferredNodeName,
+ WriteContext ctx = new WriteContext();
+ ctx.pm = forOp;
+
+ return ch.serviceAsync(opCode,
+ (opCh) -> tx0 == null || tx0.isReadOnly() || forOp == null
+ || !opCh.protocolContext().isFeatureSupported(TX_DIRECT_MAPPING) ? nullCompletedFuture()
+ : tx0.enlistFuture(opCh, ctx),
+ w -> writer.accept(schema, w, ctx),
+ r -> readSchemaAndReadData(schema, r, reader, defaultValue, responseSchemaRequired, ctx, tx0),
+ resolvePreferredNode(tx0, forOp),
+ tx0 == null ? null : tx0.nodeName(),
retryPolicyOverride,
- expectNotifications);
- }
- );
- })
-
- // Read resulting schema and the rest of the response.
- .thenCompose(t -> loadSchemaAndReadData(t, reader))
- .whenComplete((res, err) -> {
+ expectNotifications)
+ // Read resulting schema and the rest of the response.
+ .thenCompose(t -> loadSchemaAndReadData(t, reader))
+ .handle((ret, ex) -> {
+ if (ex != null) {
+ // In case of direct mapping failure try to roll back the transaction.
+ if (ctx.enlistmentToken != null && !matchAny(unwrapCause(ex), TX_ALREADY_FINISHED_ERR,
+ TX_ALREADY_FINISHED_WITH_TIMEOUT_ERR)) {
+ assert tx0 != null && !tx0.isReadOnly() : "Invalid transaction for direct mapping " + tx;
+
+ return tx0.rollbackAsync().handle((ignored, err0) -> {
+ if (err0 != null) {
+ ex.addSuppressed(err0);
+ }
+
+ sneakyThrow(ex);
+
+ return (T) null;
+ });
+ } else {
+ sneakyThrow(ex);
+ }
+ }
+
+ return completedFuture(ret);
+ }).thenCompose(identity());
+ });
+ }).whenComplete((res, err) -> {
if (err == null) {
fut.complete(res);
return;
@@ -534,13 +582,39 @@ public class ClientTable implements Table {
return fut;
}
+ private static @Nullable String resolvePreferredNode(@Nullable ClientTransaction tx, @Nullable PartitionMapping pm) {
+ String opNode = pm == null ? null : pm.nodeConsistentId();
+
+ if (tx != null) {
+ return tx.hasCommitPartition() && opNode != null ? opNode : tx.nodeName();
+ } else {
+ return opNode;
+ }
+ }
+
private <T> @Nullable Object readSchemaAndReadData(
ClientSchema knownSchema,
PayloadInputChannel in,
BiFunction<ClientSchema, PayloadInputChannel, T> fn,
@Nullable T defaultValue,
- boolean responseSchemaRequired
+ boolean responseSchemaRequired,
+ WriteContext ctx,
+ @Nullable ClientTransaction tx
) {
+ // Use enlistment meta only for remote transactions.
+ if (ctx.enlistmentToken != null) {
+ assert tx != null;
+ assert ctx.pm != null;
+
+ String consistentId = in.in().unpackString();
+ long token = in.in().unpackLong();
+
+ // Finish enlist on first request only.
+ if (ctx.enlistmentToken == 0) {
+ tx.tryFinishEnlist(ctx.pm, consistentId, token);
+ }
+ }
+
int schemaVer = in.in().unpackInt();
if (!responseSchemaRequired) {
@@ -572,7 +646,7 @@ public class ClientTable implements Table {
BiFunction<ClientSchema, PayloadInputChannel, T> fn
) {
if (!(data instanceof IgniteBiTuple)) {
- return CompletableFuture.completedFuture((T) data);
+ return completedFuture((T) data);
}
var biTuple = (IgniteBiTuple<PayloadInputChannel, Integer>) data;
@@ -608,7 +682,12 @@ public class ClientTable implements Table {
&& !pa.partitionsFut.isCompletedExceptionally();
}
- synchronized CompletableFuture<List<String>> getPartitionAssignment() {
+ /**
+ * Gets partition assignment.
+ *
+ * @return The future.
+ */
+ public synchronized CompletableFuture<List<String>> getPartitionAssignment() {
long timestamp = ch.partitionAssignmentTimestamp();
PartitionAssignment pa = partitionAssignment;
@@ -687,20 +766,13 @@ public class ClientTable implements Table {
return partitionCount;
}
- @Nullable
- private static String getPreferredNodeName(
- @Nullable PartitionAwarenessProvider provider,
+ private static @Nullable PartitionMapping getPreferredNodeName(
+ int tableId,
+ PartitionAwarenessProvider provider,
@Nullable List<String> partitions,
- ClientSchema schema) {
- if (provider == null) {
- return null;
- }
-
- String nodeName = provider.nodeName();
-
- if (nodeName != null) {
- return nodeName;
- }
+ ClientSchema schema,
+ boolean coord) {
+ assert provider != null;
if (partitions == null || partitions.isEmpty()) {
return null;
@@ -709,16 +781,25 @@ public class ClientTable implements Table {
Integer partition = provider.partition();
if (partition != null) {
- return partitions.get(partition);
+ String node = partitions.get(partition);
+ if (node == null) {
+ return null; // Mapping is incomplete.
+ }
+ return new PartitionMapping(tableId, node, partition);
}
- Integer hash = provider.getObjectHashCode(schema);
-
+ Integer hash = provider.getObjectHashCode(schema, coord);
if (hash == null) {
return null;
}
- return partitions.get(Math.abs(hash % partitions.size()));
+ int part = Math.abs(hash % partitions.size());
+
+ String node = partitions.get(part);
+ if (node == null) {
+ return null; // Mapping is incomplete.
+ }
+ return new PartitionMapping(tableId, node, part);
}
private static List<String> emptyAssignment(int size) {
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTupleSerializer.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTupleSerializer.java
index 374a1e98113..5a3e5bc08d3 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTupleSerializer.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/ClientTupleSerializer.java
@@ -25,6 +25,7 @@ import java.util.BitSet;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
@@ -32,11 +33,11 @@ import java.util.Set;
import org.apache.ignite.internal.binarytuple.BinaryTupleBuilder;
import org.apache.ignite.internal.binarytuple.BinaryTupleReader;
import org.apache.ignite.internal.client.PayloadOutputChannel;
+import org.apache.ignite.internal.client.WriteContext;
import org.apache.ignite.internal.client.proto.ClientBinaryTupleUtils;
import org.apache.ignite.internal.client.proto.ClientMessagePacker;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.TuplePart;
-import org.apache.ignite.internal.client.tx.ClientLazyTransaction;
import org.apache.ignite.internal.lang.IgniteBiTuple;
import org.apache.ignite.internal.marshaller.UnmappedColumnsException;
import org.apache.ignite.internal.util.HashCalculator;
@@ -68,14 +69,16 @@ public class ClientTupleSerializer {
* @param tuple Tuple.
* @param schema Schema.
* @param out Out.
+ * @param ctx Write context.
*/
void writeTuple(
@Nullable Transaction tx,
Tuple tuple,
ClientSchema schema,
- PayloadOutputChannel out
+ PayloadOutputChannel out,
+ WriteContext ctx
) {
- writeTuple(tx, tuple, schema, out, false, false);
+ writeTuple(tx, tuple, schema, out, ctx, false, false);
}
/**
@@ -84,6 +87,7 @@ public class ClientTupleSerializer {
* @param tuple Tuple.
* @param schema Schema.
* @param out Out.
+ * @param ctx Write context.
* @param keyOnly Key only.
*/
void writeTuple(
@@ -91,9 +95,10 @@ public class ClientTupleSerializer {
Tuple tuple,
ClientSchema schema,
PayloadOutputChannel out,
+ WriteContext ctx,
boolean keyOnly
) {
- writeTuple(tx, tuple, schema, out, keyOnly, false);
+ writeTuple(tx, tuple, schema, out, ctx, keyOnly, false);
}
/**
@@ -102,6 +107,7 @@ public class ClientTupleSerializer {
* @param tuple Tuple.
* @param schema Schema.
* @param out Out.
+ * @param ctx Write context.
* @param keyOnly Key only.
* @param skipHeader Skip header.
*/
@@ -110,12 +116,13 @@ public class ClientTupleSerializer {
Tuple tuple,
ClientSchema schema,
PayloadOutputChannel out,
+ WriteContext ctx,
boolean keyOnly,
boolean skipHeader
) {
if (!skipHeader) {
out.out().packInt(tableId);
- writeTx(tx, out);
+ writeTx(tx, out, ctx);
out.out().packInt(schema.version());
}
@@ -170,11 +177,12 @@ public class ClientTupleSerializer {
@Nullable Tuple val,
ClientSchema schema,
PayloadOutputChannel out,
+ WriteContext ctx,
boolean skipHeader
) {
if (!skipHeader) {
out.out().packInt(tableId);
- writeTx(tx, out);
+ writeTx(tx, out, ctx);
out.out().packInt(schema.version());
}
@@ -225,14 +233,20 @@ public class ClientTupleSerializer {
* @param schema Schema.
* @param out Out.
*/
- void writeKvTuples(@Nullable Transaction tx, Collection<Entry<Tuple, Tuple>> pairs, ClientSchema schema, PayloadOutputChannel out) {
+ void writeKvTuples(
+ @Nullable Transaction tx,
+ Collection<Entry<Tuple, Tuple>> pairs,
+ ClientSchema schema,
+ PayloadOutputChannel out,
+ WriteContext ctx
+ ) {
out.out().packInt(tableId);
- writeTx(tx, out);
+ writeTx(tx, out, ctx);
out.out().packInt(schema.version());
out.out().packInt(pairs.size());
for (Map.Entry<Tuple, Tuple> pair : pairs) {
- writeKvTuple(tx, pair.getKey(), pair.getValue(), schema, out, true);
+ writeKvTuple(tx, pair.getKey(), pair.getValue(), schema, out, ctx, true);
}
}
@@ -250,7 +264,8 @@ public class ClientTupleSerializer {
Collection<Entry<Tuple, Tuple>> pairs,
@Nullable BitSet deleted,
ClientSchema schema,
- PayloadOutputChannel out) {
+ PayloadOutputChannel out
+ ) {
ClientMessagePacker w = out.out();
w.packInt(tableId);
@@ -265,9 +280,9 @@ public class ClientTupleSerializer {
boolean del = deleted != null && deleted.get(i++);
if (del) {
- writeTuple(null, pair.getKey(), schema, out, true, true);
+ writeTuple(null, pair.getKey(), schema, out, null, true, true);
} else {
- writeKvTuple(null, pair.getKey(), pair.getValue(), schema, out, true);
+ writeKvTuple(null, pair.getKey(), pair.getValue(), schema, out, null, true);
}
}
}
@@ -278,6 +293,7 @@ public class ClientTupleSerializer {
* @param tuples Tuples.
* @param schema Schema.
* @param out Out.
+ * @param ctx Write context.
* @param keyOnly Key only.
*/
void writeTuples(
@@ -285,15 +301,16 @@ public class ClientTupleSerializer {
Collection<Tuple> tuples,
ClientSchema schema,
PayloadOutputChannel out,
+ WriteContext ctx,
boolean keyOnly
) {
out.out().packInt(tableId);
- writeTx(tx, out);
+ writeTx(tx, out, ctx);
out.out().packInt(schema.version());
out.out().packInt(tuples.size());
for (var tuple : tuples) {
- writeTuple(tx, tuple, schema, out, keyOnly, true);
+ writeTuple(tx, tuple, schema, out, ctx, keyOnly, true);
}
}
@@ -324,7 +341,7 @@ public class ClientTupleSerializer {
int i = 0;
for (var tuple : tuples) {
boolean keyOnly = deleted != null && deleted.get(i++);
- writeTuple(null, tuple, schema, out, keyOnly, true);
+ writeTuple(null, tuple, schema, out, null, keyOnly, true);
}
}
@@ -416,24 +433,77 @@ public class ClientTupleSerializer {
/**
* Gets partition awareness provider for the specified tuple.
*
- * @param tx Transaction.
* @param rec Tuple.
* @return Partition awareness provider.
*/
- public static PartitionAwarenessProvider getPartitionAwarenessProvider(@Nullable Transaction tx, Tuple rec) {
- return PartitionAwarenessProvider.of(ClientLazyTransaction.get(tx), schema -> getColocationHash(schema, rec));
+ public static PartitionAwarenessProvider getPartitionAwarenessProvider(Tuple rec) {
+ return PartitionAwarenessProvider.of((schema, coord) -> getColocationHash(schema, rec));
+ }
+
+ /**
+ * Gets partition awareness provider for the tuple collections.
+ *
+ * @param recs Tuples.
+ * @return Partition awareness provider.
+ */
+ public static PartitionAwarenessProvider getPartitionAwarenessProvider(Collection<Tuple> recs) {
+ return PartitionAwarenessProvider.of((schema, coord) -> {
+ Iterator<Tuple> iter = recs.iterator();
+ int hash = getColocationHash(schema, iter.next());
+
+ // For coordinator use first record for a placement hint.
+ if (coord) {
+ return hash;
+ }
+
+ while (iter.hasNext()) {
+ Tuple next = iter.next();
+ if (hash != getColocationHash(schema, next)) {
+ return null;
+ }
+ }
+
+ // If batch for the same partition can do direct enlistment. This is temporary limitation TODO ticket.
+ return hash;
+ });
}
/**
* Gets partition awareness provider for the specified object.
*
- * @param tx Transaction.
* @param rec Object.
* @return Partition awareness provider.
*/
- public static PartitionAwarenessProvider getPartitionAwarenessProvider(
- @Nullable Transaction tx, Mapper<?> mapper, Object rec) {
- return PartitionAwarenessProvider.of(ClientLazyTransaction.get(tx), schema -> getColocationHash(schema, mapper, rec));
+ public static <T> PartitionAwarenessProvider getPartitionAwarenessProvider(Mapper<?> m, T rec) {
+ return PartitionAwarenessProvider.of((schema, coord) -> getColocationHash(schema, m, rec));
+ }
+
+ /**
+ * Gets partition awareness provider for the objects collection.
+ *
+ * @param recs Objects.
+ * @return Partition awareness provider.
+ */
+ public static <T> PartitionAwarenessProvider getPartitionAwarenessProvider(Mapper<?> m, Collection<T> recs) {
+ return PartitionAwarenessProvider.of((schema, coord) -> {
+ Iterator<T> iter = recs.iterator();
+ int hash = getColocationHash(schema, m, iter.next());
+
+ // For assign coordinator mode use first record for a placement hint.
+ if (coord) {
+ return hash;
+ }
+
+ while (iter.hasNext()) {
+ Object next = iter.next();
+ if (hash != getColocationHash(schema, m, next)) {
+ return null;
+ }
+ }
+
+ // If batch for the same partition can do direct enlistment. This is temporary limitation TODO ticket.
+ return hash;
+ });
}
/**
@@ -455,7 +525,7 @@ public class ClientTupleSerializer {
return hashCalc.hash();
}
- static Integer getColocationHash(ClientSchema schema, Mapper<?> mapper, Object rec) {
+ static int getColocationHash(ClientSchema schema, Mapper<?> mapper, Object rec) {
// Colocation columns are always part of the key - https://cwiki.apache.org/confluence/display/IGNITE/IEP-86%3A+Colocation+Key.
var hashCalc = new HashCalculator();
var marsh = schema.getMarshaller(mapper, TuplePart.KEY, true);
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/table/PartitionAwarenessProvider.java b/modules/client/src/main/java/org/apache/ignite/internal/client/table/PartitionAwarenessProvider.java
index 921f7e9f9d9..cd35b23bae3 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/table/PartitionAwarenessProvider.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/table/PartitionAwarenessProvider.java
@@ -17,8 +17,7 @@
package org.apache.ignite.internal.client.table;
-import java.util.function.Function;
-import org.apache.ignite.internal.client.tx.ClientLazyTransaction;
+import java.util.function.BiFunction;
import org.jetbrains.annotations.Nullable;
/**
@@ -31,44 +30,32 @@ import org.jetbrains.annotations.Nullable;
public class PartitionAwarenessProvider {
private final @Nullable Integer partition;
- private final @Nullable Function<ClientSchema, Integer> hashFunc;
+ private final @Nullable BiFunction<ClientSchema, Boolean, Integer> hashFunc;
- private final @Nullable ClientLazyTransaction tx;
+ private PartitionAwarenessProvider(@Nullable BiFunction<ClientSchema, Boolean, Integer> hashFunc, @Nullable Integer partition) {
+ assert hashFunc != null ^ partition != null;
- private PartitionAwarenessProvider(
- @Nullable Function<ClientSchema, Integer> hashFunc,
- @Nullable Integer partition,
- @Nullable ClientLazyTransaction tx) {
this.hashFunc = hashFunc;
this.partition = partition;
- this.tx = tx;
}
public static PartitionAwarenessProvider of(Integer partition) {
- return new PartitionAwarenessProvider(null, partition, null);
+ return new PartitionAwarenessProvider(null, partition);
}
- public static PartitionAwarenessProvider of(@Nullable ClientLazyTransaction tx, Function<ClientSchema, Integer> hashFunc) {
- return new PartitionAwarenessProvider(hashFunc, null, tx);
- }
-
- @Nullable String nodeName() {
- return tx != null ? tx.nodeName() : null;
+ public static PartitionAwarenessProvider of(BiFunction<ClientSchema, Boolean, Integer> hashFunc) {
+ return new PartitionAwarenessProvider(hashFunc, null);
}
@Nullable Integer partition() {
return partition;
}
- Integer getObjectHashCode(ClientSchema schema) {
+ @Nullable Integer getObjectHashCode(ClientSchema schema, boolean coord) {
if (hashFunc == null) {
throw new IllegalStateException("Partition awareness is not enabled. Check channel() first.");
}
- return hashFunc.apply(schema);
- }
-
- boolean isPartitionAwarenessEnabled() {
- return hashFunc != null || partition != null;
+ return hashFunc.apply(schema, coord);
}
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientLazyTransaction.java b/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientLazyTransaction.java
index 6e727999d41..e3c2697d94b 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientLazyTransaction.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientLazyTransaction.java
@@ -20,7 +20,9 @@ package org.apache.ignite.internal.client.tx;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.client.PartitionMapping;
import org.apache.ignite.internal.client.ReliableChannel;
+import org.apache.ignite.internal.hlc.HybridTimestampTracker;
import org.apache.ignite.tx.Transaction;
import org.apache.ignite.tx.TransactionException;
import org.apache.ignite.tx.TransactionOptions;
@@ -30,13 +32,13 @@ import org.jetbrains.annotations.Nullable;
* Lazy client transaction. Will be actually started on the first operation.
*/
public class ClientLazyTransaction implements Transaction {
- private final long observableTimestamp;
+ private final HybridTimestampTracker observableTimestamp;
private final @Nullable TransactionOptions options;
private volatile CompletableFuture<ClientTransaction> tx;
- ClientLazyTransaction(long observableTimestamp, @Nullable TransactionOptions options) {
+ ClientLazyTransaction(HybridTimestampTracker observableTimestamp, @Nullable TransactionOptions options) {
this.observableTimestamp = observableTimestamp;
this.options = options;
}
@@ -99,13 +101,13 @@ public class ClientLazyTransaction implements Transaction {
*
* @return Node name or {@code null}.
*/
- public @Nullable String nodeName() {
+ public String nodeName() {
var tx0 = tx;
+ assert tx0 != null;
+
//noinspection resource
- return tx0 != null
- ? tx0.join().channel().protocolContext().clusterNode().name()
- : null;
+ return tx0.join().nodeName();
}
/**
@@ -132,13 +134,14 @@ public class ClientLazyTransaction implements Transaction {
*
* @param tx Transaction.
* @param ch Channel.
- * @param preferredNodeName Preferred node name.
+ * @param pm Partition mapping.
* @return Future that will be completed when the transaction is started.
*/
public static CompletableFuture<ClientTransaction> ensureStarted(
@Nullable Transaction tx,
ReliableChannel ch,
- @Nullable String preferredNodeName) {
+ @Nullable PartitionMapping pm
+ ) {
if (tx == null) {
return nullCompletedFuture();
}
@@ -147,19 +150,20 @@ public class ClientLazyTransaction implements Transaction {
throw ClientTransaction.unsupportedTxTypeException(tx);
}
- return ((ClientLazyTransaction) tx).ensureStarted(ch, preferredNodeName);
+ return ((ClientLazyTransaction) tx).ensureStarted(ch, pm);
}
private synchronized CompletableFuture<ClientTransaction> ensureStarted(
ReliableChannel ch,
- @Nullable String preferredNodeName) {
+ @Nullable PartitionMapping pm
+ ) {
var tx0 = tx;
if (tx0 != null) {
return tx0;
}
- tx0 = ClientTransactions.beginAsync(ch, preferredNodeName, options, observableTimestamp);
+ tx0 = ClientTransactions.beginAsync(ch, pm, options, observableTimestamp);
tx = tx0;
return tx0;
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransaction.java b/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransaction.java
index ade1e53aff0..73b9a8d2eb6 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransaction.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransaction.java
@@ -17,24 +17,44 @@
package org.apache.ignite.internal.client.tx;
+import static org.apache.ignite.internal.client.proto.ProtocolBitmaskFeature.TX_DIRECT_MAPPING;
import static org.apache.ignite.internal.lang.IgniteStringFormatter.format;
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.internal.util.ViewUtils.sync;
import static org.apache.ignite.lang.ErrorGroups.Common.INTERNAL_ERR;
import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_ALREADY_FINISHED_ERR;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.ignite.internal.client.ClientChannel;
+import org.apache.ignite.internal.client.PartitionMapping;
+import org.apache.ignite.internal.client.WriteContext;
import org.apache.ignite.internal.client.proto.ClientOp;
+import org.apache.ignite.internal.hlc.HybridTimestampTracker;
+import org.apache.ignite.internal.lang.IgniteBiTuple;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.tostring.IgniteToStringExclude;
+import org.apache.ignite.internal.tostring.S;
import org.apache.ignite.lang.IgniteException;
import org.apache.ignite.tx.Transaction;
import org.apache.ignite.tx.TransactionException;
+import org.jetbrains.annotations.Nullable;
+import org.jetbrains.annotations.TestOnly;
/**
* Client transaction.
*/
public class ClientTransaction implements Transaction {
+ private static final int NO_COMMIT_PARTITION = -1;
+
+ public static final UUID EMPTY = new UUID(0, 0);
+
/** Open state. */
private static final int STATE_OPEN = 0;
@@ -45,12 +65,14 @@ public class ClientTransaction implements Transaction {
private static final int STATE_ROLLED_BACK = 2;
/** Channel that the transaction belongs to. */
+ @IgniteToStringExclude
private final ClientChannel ch;
/** Transaction id. */
private final long id;
/** The future used on repeated commit/rollback. */
+ @IgniteToStringExclude
private final AtomicReference<CompletableFuture<Void>> finishFut = new AtomicReference<>();
/** State. */
@@ -59,16 +81,72 @@ public class ClientTransaction implements Transaction {
/** Read-only flag. */
private final boolean isReadOnly;
+ private final UUID txId;
+
+ private final int commitTableId;
+
+ private final int commitPartition;
+
+ private final UUID coordId;
+
+ private final String nodeName;
+
+ private final long timeout;
+
+ /** Direct enlistment map. */
+ @IgniteToStringExclude
+ private final Map<TablePartitionId, CompletableFuture<IgniteBiTuple<String, Long>>> enlisted = new ConcurrentHashMap<>();
+
+ @IgniteToStringExclude
+ private final HybridTimestampTracker tracker;
+
+ @IgniteToStringExclude
+ private final ReentrantReadWriteLock enlistPartitionLock = new ReentrantReadWriteLock();
+
/**
* Constructor.
*
* @param ch Channel that the transaction belongs to.
* @param id Transaction id.
+ * @param isReadOnly Read-only flag.
+ * @param txId Transaction id.
+ * @param cpm The commit partition mapping or {@code null} if not known.
+ * @param coordId Tx coordinator id.
+ * @param tracker The tracker.
+ * @param timeout The timeout.
*/
- public ClientTransaction(ClientChannel ch, long id, boolean isReadOnly) {
+ public ClientTransaction(
+ ClientChannel ch,
+ long id,
+ boolean isReadOnly,
+ UUID txId,
+ @Nullable PartitionMapping cpm,
+ UUID coordId,
+ HybridTimestampTracker tracker,
+ long timeout
+ ) {
this.ch = ch;
this.id = id;
this.isReadOnly = isReadOnly;
+ this.txId = txId;
+ this.nodeName = ch.protocolContext().clusterNode().name();
+ this.tracker = tracker;
+ this.timeout = timeout;
+
+ if (cpm != null) {
+ // If mapping is known, assign commit partition.
+ // However, we don't require direct connection to a commit partition primary replica here because where is a guarantee that
+ // commit partition will be assigned to provided value at the txn beginning.
+ this.commitTableId = cpm.tableId();
+ this.commitPartition = cpm.partition();
+ } else {
+ this.commitTableId = NO_COMMIT_PARTITION;
+ this.commitPartition = NO_COMMIT_PARTITION;
+ }
+
+ this.coordId = coordId;
+
+ assert this.coordId != null;
}
/**
@@ -80,6 +158,39 @@ public class ClientTransaction implements Transaction {
return id;
}
+ public UUID txId() {
+ return txId;
+ }
+
+ public int commitTableId() {
+ return commitTableId;
+ }
+
+ public int commitPartition() {
+ return commitPartition;
+ }
+
+ public boolean hasCommitPartition() {
+ return commitPartition != NO_COMMIT_PARTITION;
+ }
+
+ public UUID coordinatorId() {
+ return coordId;
+ }
+
+ public long timeout() {
+ return timeout;
+ }
+
+ /**
+ * Get coordinator node name.
+ *
+ * @return The name.
+ */
+ public String nodeName() {
+ return nodeName;
+ }
+
/**
* Gets the associated channel.
*
@@ -98,15 +209,37 @@ public class ClientTransaction implements Transaction {
/** {@inheritDoc} */
@Override
public CompletableFuture<Void> commitAsync() {
- if (!finishFut.compareAndSet(null, new CompletableFuture<>())) {
- return finishFut.get();
+ enlistPartitionLock.writeLock().lock();
+
+ try {
+ if (!finishFut.compareAndSet(null, new CompletableFuture<>())) {
+ return finishFut.get();
+ }
+ } finally {
+ enlistPartitionLock.writeLock().unlock();
}
- setState(STATE_COMMITTED);
+ CompletableFuture<Void> mainFinishFut = ch.serviceAsync(ClientOp.TX_COMMIT, w -> {
+ w.out().packLong(id);
+ if (!isReadOnly && w.clientChannel().protocolContext().isFeatureSupported(TX_DIRECT_MAPPING)) {
+ w.out().packLong(tracker.get().longValue());
+ w.out().packInt(enlisted.size());
+ for (Entry<TablePartitionId, CompletableFuture<IgniteBiTuple<String, Long>>> entry : enlisted.entrySet()) {
+ w.out().packInt(entry.getKey().tableId());
+ w.out().packInt(entry.getKey().partitionId());
+ w.out().packString(entry.getValue().getNow(null).get1());
+ w.out().packLong(entry.getValue().getNow(null).get2());
+ }
+ }
+ }, r -> null);
- CompletableFuture<Void> mainFinishFut = ch.serviceAsync(ClientOp.TX_COMMIT, w -> w.out().packLong(id), r -> null);
+ mainFinishFut.handle((res, e) -> {
+ setState(STATE_COMMITTED);
- mainFinishFut.handle((res, e) -> finishFut.get().complete(null));
+ finishFut.get().complete(null);
+
+ return null;
+ });
return mainFinishFut;
}
@@ -124,11 +257,26 @@ public class ClientTransaction implements Transaction {
return finishFut.get();
}
- setState(STATE_ROLLED_BACK);
+ CompletableFuture<Void> mainFinishFut = ch.serviceAsync(ClientOp.TX_ROLLBACK, w -> {
+ w.out().packLong(id);
+ if (!isReadOnly && w.clientChannel().protocolContext().isFeatureSupported(TX_DIRECT_MAPPING)) {
+ w.out().packInt(enlisted.size());
+ for (Entry<TablePartitionId, CompletableFuture<IgniteBiTuple<String, Long>>> entry : enlisted.entrySet()) {
+ w.out().packInt(entry.getKey().tableId());
+ w.out().packInt(entry.getKey().partitionId());
+ w.out().packString(entry.getValue().getNow(null).get1());
+ w.out().packLong(entry.getValue().getNow(null).get2());
+ }
+ }
+ }, r -> null);
+
+ mainFinishFut.handle((res, e) -> {
+ setState(STATE_ROLLED_BACK);
- CompletableFuture<Void> mainFinishFut = ch.serviceAsync(ClientOp.TX_ROLLBACK, w -> w.out().packLong(id), r -> null);
+ finishFut.get().complete(null);
- mainFinishFut.handle((res, e) -> finishFut.get().complete(null));
+ return null;
+ });
return mainFinishFut;
}
@@ -140,8 +288,8 @@ public class ClientTransaction implements Transaction {
}
/**
- * Gets the internal transaction from the given public transaction. Throws an exception if the given transaction is
- * not an instance of {@link ClientTransaction}.
+ * Gets the internal transaction from the given public transaction. Throws an exception if the given transaction is not an instance of
+ * {@link ClientTransaction}.
*
* @param tx Public transaction.
* @return Internal transaction.
@@ -159,12 +307,9 @@ public class ClientTransaction implements Transaction {
return clientTx;
}
- // Match org.apache.ignite.internal.tx.TxState enum:
- String stateStr = state == STATE_COMMITTED ? "COMMITTED" : "ABORTED";
-
throw new TransactionException(
TX_ALREADY_FINISHED_ERR,
- format("Transaction is already finished [id={}, state={}].", clientTx.id, stateStr));
+ format("Transaction is already finished [tx={}].", clientTx));
}
static IgniteException unsupportedTxTypeException(Transaction tx) {
@@ -176,4 +321,88 @@ public class ClientTransaction implements Transaction {
private void setState(int state) {
this.state.compareAndExchange(STATE_OPEN, state);
}
+
+ private void checkEnlistPossible() {
+ if (finishFut.get() != null) {
+ throw new TransactionException(TX_ALREADY_FINISHED_ERR, format("Transaction is already finished [tx={}].", this));
+ }
+ }
+
+ /**
+ * Enlists a write operation in direct mapping.
+ *
+ * @param opChannel Operation channge.
+ * @param ctx The context.
+ *
+ * @return The future.
+ */
+ public CompletableFuture<Void> enlistFuture(ClientChannel opChannel, WriteContext ctx) {
+ // Check if direct mapping is applicable.
+ if (ctx.pm != null && ctx.pm.nodeConsistentId().equals(opChannel.protocolContext().clusterNode().name()) && hasCommitPartition()) {
+ if (!enlistPartitionLock.readLock().tryLock()) {
+ throw new TransactionException(TX_ALREADY_FINISHED_ERR, format("Transaction is already finished [tx={}].", this));
+ }
+
+ checkEnlistPossible();
+
+ boolean[] first = {false};
+
+ TablePartitionId tablePartitionId = new TablePartitionId(ctx.pm.tableId(), ctx.pm.partition());
+
+ CompletableFuture<IgniteBiTuple<String, Long>> fut = enlisted.compute(tablePartitionId, (k, v) -> {
+ if (v == null) {
+ first[0] = true;
+ return new CompletableFuture<>();
+ } else {
+ return v;
+ }
+ });
+
+ enlistPartitionLock.readLock().unlock();
+
+ // Re-check after unlock.
+ checkEnlistPossible();
+
+ if (first[0]) {
+ ctx.enlistmentToken = 0L;
+ // For the first request return completed future.
+ return nullCompletedFuture();
+ } else {
+ return fut.thenAccept(tup -> ctx.enlistmentToken = tup.get2());
+ }
+ }
+
+ return nullCompletedFuture();
+ }
+
+ /**
+ * Tries to finish existing enlistment.
+ *
+ * @param pm Partition mapping.
+ * @param consistentId Consistent id.
+ * @param token Enlistment token.
+ */
+ public void tryFinishEnlist(PartitionMapping pm, String consistentId, long token) {
+ if (!hasCommitPartition()) {
+ return;
+ }
+
+ TablePartitionId tablePartitionId = new TablePartitionId(pm.tableId(), pm.partition());
+
+ CompletableFuture<IgniteBiTuple<String, Long>> fut = enlisted.get(tablePartitionId);
+
+ if (fut != null && !fut.isDone()) {
+ fut.complete(new IgniteBiTuple<>(consistentId, token));
+ }
+ }
+
+ @TestOnly
+ public int enlistedCount() {
+ return enlisted.size();
+ }
+
+ @Override
+ public String toString() {
+ return S.toString(this);
+ }
}
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransactions.java b/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransactions.java
index 4e027af4b4e..23bbc1f8b10 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransactions.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/tx/ClientTransactions.java
@@ -17,13 +17,18 @@
package org.apache.ignite.internal.client.tx;
+import static org.apache.ignite.internal.client.proto.ProtocolBitmaskFeature.TX_DIRECT_MAPPING;
+import static org.apache.ignite.internal.client.tx.ClientTransaction.EMPTY;
import static org.apache.ignite.internal.util.ViewUtils.sync;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.client.PartitionMapping;
import org.apache.ignite.internal.client.PayloadInputChannel;
import org.apache.ignite.internal.client.ReliableChannel;
import org.apache.ignite.internal.client.proto.ClientMessageUnpacker;
import org.apache.ignite.internal.client.proto.ClientOp;
+import org.apache.ignite.internal.hlc.HybridTimestampTracker;
import org.apache.ignite.tx.IgniteTransactions;
import org.apache.ignite.tx.Transaction;
import org.apache.ignite.tx.TransactionOptions;
@@ -62,9 +67,10 @@ public class ClientTransactions implements IgniteTransactions {
static CompletableFuture<ClientTransaction> beginAsync(
ReliableChannel ch,
- @Nullable String preferredNodeName,
+ @Nullable PartitionMapping pm,
@Nullable TransactionOptions options,
- long observableTimestamp) {
+ HybridTimestampTracker observableTimestamp
+ ) {
boolean readOnly = options != null && options.readOnly();
long timeout = options == null ? USE_CONFIGURED_TIMEOUT_DEFAULT : options.timeoutMillis();
@@ -73,19 +79,35 @@ public class ClientTransactions implements IgniteTransactions {
w -> {
w.out().packBoolean(readOnly);
w.out().packLong(timeout);
- w.out().packLong(observableTimestamp);
+ w.out().packLong(observableTimestamp.get().longValue());
+ if (!readOnly && w.clientChannel().protocolContext().isFeatureSupported(TX_DIRECT_MAPPING)) {
+ w.out().packInt(pm == null ? -1 : pm.tableId());
+ w.out().packInt(pm == null ? -1 : pm.partition());
+ }
},
- r -> readTx(r, readOnly),
- preferredNodeName,
+ r -> readTx(r, readOnly, pm, observableTimestamp, timeout),
+ pm == null ? null : pm.nodeConsistentId(),
+ null,
null,
false);
}
- private static ClientTransaction readTx(PayloadInputChannel r, boolean isReadOnly) {
+ private static ClientTransaction readTx(
+ PayloadInputChannel r,
+ boolean isReadOnly,
+ @Nullable PartitionMapping pm,
+ HybridTimestampTracker tracker,
+ long timeout
+ ) {
ClientMessageUnpacker in = r.in();
long id = in.unpackLong();
-
- return new ClientTransaction(r.clientChannel(), id, isReadOnly);
+ if (isReadOnly || !r.clientChannel().protocolContext().isFeatureSupported(TX_DIRECT_MAPPING)) {
+ return new ClientTransaction(r.clientChannel(), id, isReadOnly, EMPTY, null, EMPTY, null, timeout);
+ } else {
+ UUID txId = in.unpackUuid();
+ UUID coordId = in.unpackUuid();
+ return new ClientTransaction(r.clientChannel(), id, isReadOnly, txId, pm, coordId, tracker, timeout);
+ }
}
}
diff --git a/modules/client/src/test/java/org/apache/ignite/client/ObservableTimestampPropagationTest.java b/modules/client/src/test/java/org/apache/ignite/client/ObservableTimestampPropagationTest.java
index 102e42e82a1..09275b81a49 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/ObservableTimestampPropagationTest.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/ObservableTimestampPropagationTest.java
@@ -30,7 +30,6 @@ import org.apache.ignite.internal.client.TcpIgniteClient;
import org.apache.ignite.internal.client.tx.ClientLazyTransaction;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.tx.TransactionOptions;
-import org.jetbrains.annotations.Nullable;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -70,7 +69,7 @@ public class ObservableTimestampPropagationTest extends BaseIgniteAbstractTest {
// +1 because logical time is incremented on every call to nowLong - on client handler start.
assertEquals(
(currentServerTimestamp.get() << LOGICAL_TIME_BITS_SIZE) + 1,
- ch.observableTimestamp(),
+ ch.observableTimestamp().get().longValue(),
"Handshake should initialize observable timestamp");
// RW TX does not propagate timestamp.
@@ -99,7 +98,7 @@ public class ObservableTimestampPropagationTest extends BaseIgniteAbstractTest {
assertEquals(11, lastObservableTimestamp(ch));
}
- private static @Nullable Long lastObservableTimestamp(ReliableChannel ch) {
- return ch.observableTimestamp() >> LOGICAL_TIME_BITS_SIZE;
+ private static Long lastObservableTimestamp(ReliableChannel ch) {
+ return ch.observableTimestamp().get().longValue() >> LOGICAL_TIME_BITS_SIZE;
}
}
diff --git a/modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java b/modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java
index 8c7cd103117..b0c85667e29 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/TestClientHandlerModule.java
@@ -51,6 +51,7 @@ import org.apache.ignite.client.handler.configuration.ClientConnectorConfigurati
import org.apache.ignite.internal.catalog.CatalogService;
import org.apache.ignite.internal.client.proto.ClientMessageDecoder;
import org.apache.ignite.internal.client.proto.HandshakeExtension;
+import org.apache.ignite.internal.client.proto.ProtocolBitmaskFeature;
import org.apache.ignite.internal.compute.IgniteComputeInternal;
import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.hlc.TestClockService;
@@ -211,6 +212,9 @@ public class TestClientHandlerModule implements IgniteComponent {
ServerBootstrap bootstrap = bootstrapFactory.createServerBootstrap();
+ BitSet features = BitSet.valueOf(new long[]{ThreadLocalRandom.current().nextLong()});
+ features.set(ProtocolBitmaskFeature.TX_DIRECT_MAPPING.featureId());
+
bootstrap.childHandler(new ChannelInitializer<>() {
@Override
protected void initChannel(Channel ch) {
@@ -242,7 +246,7 @@ public class TestClientHandlerModule implements IgniteComponent {
new TestLowWatermark()
),
Runnable::run,
- BitSet.valueOf(new long[]{ThreadLocalRandom.current().nextLong()}),
+ features,
randomExtensions(),
Runnable::run
)
diff --git a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java
index 6554c2be9fb..7ad40fbdf31 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeInternalTable.java
@@ -546,6 +546,11 @@ public class FakeInternalTable implements InternalTable, StreamerReceiverRunner
return this;
}
+ @Override
+ public ReplicationGroupId targetReplicationGroupId(int partId) {
+ return null; // Not supported yet.
+ }
+
@Override
public <A, I, R> CompletableFuture<Collection<R>> runReceiverAsync(ReceiverDescriptor<A> receiver, @Nullable A receiverArg,
Collection<I> items, ClusterNode node, List<DeploymentUnit> deploymentUnits) {
diff --git a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeTxManager.java b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeTxManager.java
index 4a3d0ee3088..bde7c42df4e 100644
--- a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeTxManager.java
+++ b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeTxManager.java
@@ -142,7 +142,7 @@ public class FakeTxManager implements TxManager {
@Override
public UUID coordinatorId() {
- return null;
+ return id;
}
@Override
@@ -177,11 +177,6 @@ public class FakeTxManager implements TxManager {
return 10_000;
}
- @Override
- public long getTimeoutOrDefault(long defaultTimeout) {
- return 10_000;
- }
-
@Override
public CompletableFuture<Void> kill() {
return nullCompletedFuture();
@@ -284,4 +279,10 @@ public class FakeTxManager implements TxManager {
) {
// No-op.
}
+
+ @Override
+ public InternalTransaction beginRemote(UUID txId, TablePartitionId commitPartId, UUID coord, long token,
+ long timeout) {
+ return null;
+ }
}
diff --git a/modules/client/src/test/java/org/apache/ignite/internal/client/RepeatedFinishClientTransactionTest.java b/modules/client/src/test/java/org/apache/ignite/internal/client/RepeatedFinishClientTransactionTest.java
index 2d42de49cfa..813d5e840c7 100644
--- a/modules/client/src/test/java/org/apache/ignite/internal/client/RepeatedFinishClientTransactionTest.java
+++ b/modules/client/src/test/java/org/apache/ignite/internal/client/RepeatedFinishClientTransactionTest.java
@@ -17,8 +17,11 @@
package org.apache.ignite.internal.client;
+import static java.util.UUID.randomUUID;
import static java.util.concurrent.CompletableFuture.failedFuture;
+import static org.apache.ignite.internal.hlc.HybridTimestampTracker.EMPTY_TS_PROVIDER;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotSame;
import static org.junit.jupiter.api.Assertions.assertSame;
@@ -34,7 +37,10 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.ignite.internal.client.tx.ClientTransaction;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.apache.ignite.lang.ErrorGroups.Transactions;
+import org.apache.ignite.tx.TransactionException;
import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
/**
* Tests repeated commit/rollback operations.
@@ -45,15 +51,23 @@ public class RepeatedFinishClientTransactionTest extends BaseIgniteAbstractTest
CountDownLatch txFinishStartedLatch = new CountDownLatch(1);
CountDownLatch secondFinishLatch = new CountDownLatch(1);
- TestClientChannel clientChannel = new TestClientChannel(txFinishStartedLatch, secondFinishLatch);
+ ProtocolContext ctx = mock(ProtocolContext.class, Mockito.RETURNS_DEEP_STUBS);
+ when(ctx.clusterNode()).thenReturn(new ClientClusterNode(randomUUID(), "test", null));
- ClientTransaction tx = new ClientTransaction(clientChannel, 1, false);
+ TestClientChannel clientChannel = new TestClientChannel(txFinishStartedLatch, secondFinishLatch) {
+ @Override
+ public ProtocolContext protocolContext() {
+ return ctx;
+ }
+ };
+
+ ClientTransaction tx = new ClientTransaction(clientChannel, 1, false, randomUUID(), null, randomUUID(), EMPTY_TS_PROVIDER, 0);
CompletableFuture<Object> fut = new CompletableFuture<>();
CompletableFuture<Void> firstCommitFut = fut.thenComposeAsync((ignored) -> tx.commitAsync());
- fut.complete(null);
+ fut.complete(null); // commitAsync will be called from common pool.
txFinishStartedLatch.await();
@@ -83,9 +97,17 @@ public class RepeatedFinishClientTransactionTest extends BaseIgniteAbstractTest
CountDownLatch txFinishStartedLatch = new CountDownLatch(1);
CountDownLatch secondFinishLatch = new CountDownLatch(1);
- TestClientChannel clientChannel = new TestClientChannel(txFinishStartedLatch, secondFinishLatch);
+ ProtocolContext ctx = mock(ProtocolContext.class, Mockito.RETURNS_DEEP_STUBS);
+ when(ctx.clusterNode()).thenReturn(new ClientClusterNode(randomUUID(), "test", null));
+
+ TestClientChannel clientChannel = new TestClientChannel(txFinishStartedLatch, secondFinishLatch) {
+ @Override
+ public ProtocolContext protocolContext() {
+ return ctx;
+ }
+ };
- ClientTransaction tx = new ClientTransaction(clientChannel, 1, false);
+ ClientTransaction tx = new ClientTransaction(clientChannel, 1, false, randomUUID(), null, randomUUID(), EMPTY_TS_PROVIDER, 0);
CompletableFuture<Object> fut = new CompletableFuture<>();
@@ -118,11 +140,15 @@ public class RepeatedFinishClientTransactionTest extends BaseIgniteAbstractTest
@Test
public void testRepeatedCommitRollbackAfterCommitWithException() throws Exception {
- TestClientChannel clientChannel = mock(TestClientChannel.class);
+ TestClientChannel clientChannel = mock(TestClientChannel.class, Mockito.RETURNS_DEEP_STUBS);
+ ProtocolContext ctx = mock(ProtocolContext.class, Mockito.RETURNS_DEEP_STUBS);
+ when(ctx.clusterNode()).thenReturn(new ClientClusterNode(randomUUID(), "test", null));
+
+ when(clientChannel.protocolContext()).thenReturn(ctx);
when(clientChannel.serviceAsync(anyInt(), any(), any())).thenReturn(failedFuture(new Exception("Expected exception.")));
- ClientTransaction tx = new ClientTransaction(clientChannel, 1, false);
+ ClientTransaction tx = new ClientTransaction(clientChannel, 1, false, randomUUID(), null, randomUUID(), EMPTY_TS_PROVIDER, 0);
CompletableFuture<Object> fut = new CompletableFuture<>();
@@ -144,11 +170,15 @@ public class RepeatedFinishClientTransactionTest extends BaseIgniteAbstractTest
@Test
public void testRepeatedCommitRollbackAfterRollbackWithException() throws Exception {
- TestClientChannel clientChannel = mock(TestClientChannel.class);
+ TestClientChannel clientChannel = mock(TestClientChannel.class, Mockito.RETURNS_DEEP_STUBS);
+
+ ProtocolContext ctx = mock(ProtocolContext.class, Mockito.RETURNS_DEEP_STUBS);
+ when(ctx.clusterNode()).thenReturn(new ClientClusterNode(randomUUID(), "test", null));
+ when(clientChannel.protocolContext()).thenReturn(ctx);
when(clientChannel.serviceAsync(anyInt(), any(), any())).thenReturn(failedFuture(new Exception("Expected exception.")));
- ClientTransaction tx = new ClientTransaction(clientChannel, 1, false);
+ ClientTransaction tx = new ClientTransaction(clientChannel, 1, false, randomUUID(), null, randomUUID(), EMPTY_TS_PROVIDER, 0);
CompletableFuture<Object> fut = new CompletableFuture<>();
@@ -168,6 +198,62 @@ public class RepeatedFinishClientTransactionTest extends BaseIgniteAbstractTest
tx.rollbackAsync().get(3, TimeUnit.SECONDS);
}
+ @Test
+ public void testEnlistFailAfterCommit() {
+ TestClientChannel clientChannel = mock(TestClientChannel.class, Mockito.RETURNS_DEEP_STUBS);
+
+ ProtocolContext ctx = mock(ProtocolContext.class, Mockito.RETURNS_DEEP_STUBS);
+ when(ctx.clusterNode()).thenReturn(new ClientClusterNode(randomUUID(), "test", null));
+
+ when(clientChannel.protocolContext()).thenReturn(ctx);
+ when(clientChannel.serviceAsync(anyInt(), any(), any())).thenReturn(nullCompletedFuture());
+
+ PartitionMapping pm = new PartitionMapping(1, "test", 1);
+
+ ClientTransaction tx = new ClientTransaction(clientChannel, 1, false, randomUUID(), pm, randomUUID(), EMPTY_TS_PROVIDER, 0);
+
+ tx.commit();
+
+ WriteContext wc = new WriteContext();
+ wc.pm = pm;
+
+ try {
+ tx.enlistFuture(clientChannel, wc);
+
+ fail();
+ } catch (TransactionException e) {
+ assertEquals(Transactions.TX_ALREADY_FINISHED_ERR, e.code());
+ }
+ }
+
+ @Test
+ public void testEnlistFailAfterRollback() {
+ TestClientChannel clientChannel = mock(TestClientChannel.class, Mockito.RETURNS_DEEP_STUBS);
+
+ ProtocolContext ctx = mock(ProtocolContext.class, Mockito.RETURNS_DEEP_STUBS);
+ when(ctx.clusterNode()).thenReturn(new ClientClusterNode(randomUUID(), "test", null));
+
+ when(clientChannel.protocolContext()).thenReturn(ctx);
+ when(clientChannel.serviceAsync(anyInt(), any(), any())).thenReturn(nullCompletedFuture());
+
+ PartitionMapping pm = new PartitionMapping(1, "test", 1);
+
+ ClientTransaction tx = new ClientTransaction(clientChannel, 1, false, randomUUID(), pm, randomUUID(), EMPTY_TS_PROVIDER, 0);
+
+ tx.rollback();
+
+ WriteContext wc = new WriteContext();
+ wc.pm = pm;
+
+ try {
+ tx.enlistFuture(clientChannel, wc);
+
+ fail();
+ } catch (TransactionException e) {
+ assertEquals(Transactions.TX_ALREADY_FINISHED_ERR, e.code());
+ }
+ }
+
private static class TestClientChannel implements ClientChannel {
private final CountDownLatch txFinishStartedLatch;
diff --git a/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcConnection.java b/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcConnection.java
index 86bce833651..4a43aeee1e2 100644
--- a/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcConnection.java
+++ b/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcConnection.java
@@ -868,7 +868,7 @@ public class JdbcConnection implements Connection {
/** Returns the latest time observed by client. */
long observableTimestamp() {
- return client.channel().observableTimestamp();
+ return client.channel().observableTimestamp().get().longValue();
}
/** {@inheritDoc} */
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSingleRowPkReplicaRequest.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSingleRowPkReplicaRequest.java
index e0c13a3aecd..7a8642849c2 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSingleRowPkReplicaRequest.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSingleRowPkReplicaRequest.java
@@ -26,4 +26,10 @@ import org.apache.ignite.internal.replicator.message.TableAware;
*/
@Transferable(PartitionReplicationMessageGroup.RW_SINGLE_ROW_PK_REPLICA_REQUEST)
public interface ReadWriteSingleRowPkReplicaRequest extends SingleRowPkReplicaRequest, ReadWriteReplicaRequest, TableAware {
+ /**
+ * Disable delayed ack optimization.
+ *
+ * @return {@code True} to disable the delayed ack optimization.
+ */
+ boolean skipDelayedAck();
}
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSingleRowReplicaRequest.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSingleRowReplicaRequest.java
index 4f33af949a0..0d3993ce1e5 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSingleRowReplicaRequest.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSingleRowReplicaRequest.java
@@ -26,4 +26,10 @@ import org.apache.ignite.internal.replicator.message.TableAware;
*/
@Transferable(PartitionReplicationMessageGroup.RW_SINGLE_ROW_REPLICA_REQUEST)
public interface ReadWriteSingleRowReplicaRequest extends SingleRowReplicaRequest, ReadWriteReplicaRequest, TableAware {
+ /**
+ * Disable delayed ack optimization.
+ *
+ * @return {@code True} to disable the delayed ack optimization.
+ */
+ boolean skipDelayedAck();
}
diff --git a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSwapRowReplicaRequest.java b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSwapRowReplicaRequest.java
index 65ee84ac863..8e2a7a5dff4 100644
--- a/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSwapRowReplicaRequest.java
+++ b/modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/network/replication/ReadWriteSwapRowReplicaRequest.java
@@ -26,5 +26,10 @@ import org.apache.ignite.internal.replicator.message.TableAware;
*/
@Transferable(PartitionReplicationMessageGroup.RW_DUAL_ROW_REPLICA_REQUEST)
public interface ReadWriteSwapRowReplicaRequest extends SwapRowReplicaRequest, ReadWriteReplicaRequest, TableAware {
-
+ /**
+ * Disable delayed ack optimization.
+ *
+ * @return {@code True} to disable the delayed ack optimization.
+ */
+ boolean skipDelayedAck();
}
diff --git a/modules/platforms/cpp/ignite/odbc/sql_connection.cpp b/modules/platforms/cpp/ignite/odbc/sql_connection.cpp
index ea707d48799..977d465681f 100644
--- a/modules/platforms/cpp/ignite/odbc/sql_connection.cpp
+++ b/modules/platforms/cpp/ignite/odbc/sql_connection.cpp
@@ -434,6 +434,7 @@ void sql_connection::transaction_start() {
sync_request(protocol::client_operation::TX_BEGIN, [&](protocol::writer &writer) {
writer.write_bool(false); // read_only.
writer.write(std::int64_t(0)); // timeoutMillis.
+ writer.write(get_observable_timestamp());
});
protocol::reader reader(response.get_bytes_view());
diff --git a/modules/runner/build.gradle b/modules/runner/build.gradle
index 69efec98d94..9bfe0a4345a 100644
--- a/modules/runner/build.gradle
+++ b/modules/runner/build.gradle
@@ -258,13 +258,31 @@ tasks.register("runnerPlatformBenchmark", JavaExec) {
tasks.register("runUpsertBenchmark", JavaExec) {
mainClass = "org.apache.ignite.internal.benchmark.UpsertKvBenchmark"
- jvmArgs += defaultJvmArgs
+ jvmArgs += addOpens + ["-Dio.netty.tryReflectionSetAccessible=true", "-Xmx16g"]
classpath = sourceSets.integrationTest.runtimeClasspath
enableAssertions = true
}
+tasks.register("runClientBenchmark", JavaExec) {
+ mainClass = "org.apache.ignite.internal.benchmark.ClientKvBenchmark"
+
+ jvmArgs += addOpens + ["-Dio.netty.tryReflectionSetAccessible=true", "-Xmx16g"]
+
+ classpath = sourceSets.integrationTest.runtimeClasspath
+
+ enableAssertions = true
+}
+
+tasks.register("runRemoteBenchmark", JavaExec) {
+ mainClass = "org.apache.ignite.internal.benchmark.RemoteKvBenchmark"
+
+ jvmArgs += addOpens + ["-Dio.netty.tryReflectionSetAccessible=true", "-Xmx16g"]
+
+ classpath = sourceSets.integrationTest.runtimeClasspath
+}
+
jar {
manifest {
attributes(
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/AbstractMultiNodeBenchmark.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/AbstractMultiNodeBenchmark.java
index 300e6f5d0f6..7baf2cc8e61 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/AbstractMultiNodeBenchmark.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/AbstractMultiNodeBenchmark.java
@@ -86,7 +86,9 @@ public class AbstractMultiNodeBenchmark {
@Setup
public void nodeSetUp() throws Exception {
System.setProperty("jraft.available_processors", "2");
- startCluster();
+ if (!remote()) {
+ startCluster();
+ }
try {
// Create a new zone on the cluster's start-up.
@@ -135,7 +137,7 @@ public class AbstractMultiNodeBenchmark {
}
protected static void createTable(String tableName, List<String> columns, List<String> primaryKeys, List<String> colocationKeys) {
- var createTableStatement = "CREATE TABLE " + tableName + "(\n";
+ var createTableStatement = "CREATE TABLE IF NOT EXISTS " + tableName + "(\n";
createTableStatement += String.join(",\n", columns);
createTableStatement += "\n, PRIMARY KEY (" + String.join(", ", primaryKeys) + ")\n)";
@@ -184,11 +186,15 @@ public class AbstractMultiNodeBenchmark {
* @throws Exception In case of any error.
*/
@TearDown
- public final void nodeTearDown() throws Exception {
+ public void nodeTearDown() throws Exception {
IgniteUtils.closeAll(igniteServers.stream().map(node -> node::shutdown));
}
private void startCluster() throws Exception {
+ if (remote()) {
+ throw new AssertionError("Can't start the cluster in remote mode");
+ }
+
Path workDir = workDir();
String connectNodeAddr = "\"localhost:" + BASE_PORT + '\"';
@@ -206,6 +212,7 @@ public class AbstractMultiNodeBenchmark {
+ " " + DEFAULT_STORAGE_PROFILE + ".size: 2073741824 " // Avoid page replacement.
+ " },\n"
+ " clientConnector: { port:{} },\n"
+ + " clientConnector.sendServerExceptionStackTraceToClient: true\n"
+ " rest.port: {},\n"
+ " raft.fsync = " + fsync() + ",\n"
+ " system.partitionsLogPath = \"" + logPath() + "\",\n"
@@ -273,4 +280,8 @@ public class AbstractMultiNodeBenchmark {
protected int replicaCount() {
return CatalogUtils.DEFAULT_REPLICA_COUNT;
}
+
+ protected boolean remote() {
+ return false;
+ }
}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/ClientKvBenchmark.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/ClientKvBenchmark.java
new file mode 100644
index 00000000000..7ee43eb4574
--- /dev/null
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/ClientKvBenchmark.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.benchmark;
+
+import static org.apache.ignite.internal.util.IgniteUtils.closeAll;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.internal.client.table.ClientTable;
+import org.apache.ignite.internal.lang.IgniteSystemProperties;
+import org.apache.ignite.table.KeyValueView;
+import org.apache.ignite.table.Tuple;
+import org.apache.ignite.tx.Transaction;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/**
+ * Benchmark for a single upsert operation via KV API with a possibility to disable updates via RAFT and to storage using a remote client.
+ */
+@State(Scope.Benchmark)
+@Fork(1)
+@Threads(4)
+@Warmup(iterations = 10, time = 2)
+@Measurement(iterations = 20, time = 2)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+public class ClientKvBenchmark extends AbstractMultiNodeBenchmark {
+ private final Tuple tuple = Tuple.create();
+
+ protected IgniteClient client;
+
+ private KeyValueView<Tuple, Tuple> kvView;
+
+ @Param({"0"})
+ private int offset; // 1073741824 for second client to ensure unique keys
+
+ @Param({"5"})
+ private int batch;
+
+ @Param({"false"})
+ private boolean fsync;
+
+ @Param({"16"})
+ private int partitionCount;
+
+ private final AtomicInteger counter = new AtomicInteger();
+
+ private final ThreadLocal<Integer> gen = ThreadLocal.withInitial(() -> offset + counter.getAndIncrement() * 20_000_000);
+
+ protected String[] addresses() {
+ return new String[]{"127.0.0.1:10800", "127.0.0.1:10801"};
+ }
+
+ @Override
+ public void nodeSetUp() throws Exception {
+ System.setProperty(IgniteSystemProperties.IGNITE_SKIP_REPLICATION_IN_BENCHMARK, "false");
+ System.setProperty(IgniteSystemProperties.IGNITE_SKIP_STORAGE_UPDATE_IN_BENCHMARK, "false");
+ super.nodeSetUp();
+ }
+
+ /**
+ * Initializes the tuple.
+ */
+ @Setup
+ public void setUp() {
+ for (int i = 1; i < 11; i++) {
+ tuple.set("field" + i, FIELD_VAL);
+ }
+
+ client = IgniteClient.builder().addresses(addresses()).build();
+ ClientTable table = (ClientTable) client.tables().table(TABLE_NAME);
+ kvView = table.keyValueView();
+ }
+
+ @Override
+ public void nodeTearDown() throws Exception {
+ closeAll(client);
+ super.nodeTearDown();
+ }
+
+ /**
+ * Benchmark for KV upsert via embedded client.
+ */
+ @Benchmark
+ public void upsert() {
+ Transaction tx = client.transactions().begin();
+ for (int i = 0; i < batch; i++) {
+ Tuple key = Tuple.create().set("ycsb_key", nextId());
+ kvView.put(tx, key, tuple);
+ }
+ tx.commit();
+ }
+
+ private int nextId() {
+ int cur = gen.get() + 1;
+ gen.set(cur);
+ return cur;
+ }
+
+ /**
+ * Benchmark's entry point.
+ */
+ public static void main(String[] args) throws RunnerException {
+ Options opt = new OptionsBuilder()
+ .include(".*" + ClientKvBenchmark.class.getSimpleName() + ".*")
+ // .jvmArgsAppend("-Djmh.executor=VIRTUAL")
+ // .addProfiler(JavaFlightRecorderProfiler.class, "configName=profile.jfc")
+ .build();
+
+ new Runner(opt).run();
+ }
+
+ @Override
+ protected boolean fsync() {
+ return fsync;
+ }
+
+ @Override
+ protected int nodes() {
+ return 2;
+ }
+
+ @Override
+ protected int partitionCount() {
+ return partitionCount;
+ }
+
+ @Override
+ protected int replicaCount() {
+ return 1;
+ }
+}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/RemoteKvBenchmark.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/RemoteKvBenchmark.java
new file mode 100644
index 00000000000..1cf5d761036
--- /dev/null
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/benchmark/RemoteKvBenchmark.java
@@ -0,0 +1,81 @@
+/*
+ * 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.benchmark;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.internal.lang.IgniteSystemProperties;
+import org.openjdk.jmh.annotations.BenchmarkMode;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.Mode;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+import org.openjdk.jmh.runner.Runner;
+import org.openjdk.jmh.runner.RunnerException;
+import org.openjdk.jmh.runner.options.Options;
+import org.openjdk.jmh.runner.options.OptionsBuilder;
+
+/**
+ * Benchmark for a single upsert operation using externally enabled cluster.
+ */
+@State(Scope.Benchmark)
+@Fork(1)
+@Threads(64)
+@Warmup(iterations = 10, time = 2)
+@Measurement(iterations = 20, time = 2)
+@BenchmarkMode(Mode.Throughput)
+@OutputTimeUnit(TimeUnit.SECONDS)
+public class RemoteKvBenchmark extends ClientKvBenchmark {
+ @Override
+ protected boolean remote() {
+ return true;
+ }
+
+ @Override
+ protected String[] addresses() {
+ return new String[]{};
+ }
+
+ @Override
+ public void nodeSetUp() throws Exception {
+ System.setProperty(IgniteSystemProperties.IGNITE_SKIP_REPLICATION_IN_BENCHMARK, "false");
+ System.setProperty(IgniteSystemProperties.IGNITE_SKIP_STORAGE_UPDATE_IN_BENCHMARK, "false");
+
+ client = IgniteClient.builder().addresses(addresses()).build();
+ publicIgnite = client;
+
+ super.nodeSetUp();
+ }
+
+ /**
+ * Benchmark's entry point.
+ */
+ public static void main(String[] args) throws RunnerException {
+ Options opt = new OptionsBuilder()
+ .include(".*" + RemoteKvBenchmark.class.getSimpleName() + ".*")
+ // .jvmArgsAppend("-Djmh.executor=VIRTUAL")
+ // .addProfiler(JavaFlightRecorderProfiler.class, "configName=profile.jfc")
+ .build();
+
+ new Runner(opt).run();
+ }
+}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/client/ItClientDirectMappingTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/client/ItClientDirectMappingTest.java
new file mode 100644
index 00000000000..2b9c28cc44f
--- /dev/null
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/client/ItClientDirectMappingTest.java
@@ -0,0 +1,150 @@
+/*
+ * 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.client;
+
+import static org.apache.ignite.internal.TestDefaultProfilesNames.DEFAULT_AIPERSIST_PROFILE_NAME;
+import static org.apache.ignite.internal.sql.engine.util.SqlTestUtils.executeUpdate;
+import static org.apache.ignite.internal.tx.test.ItTransactionTestUtils.withTx;
+import static org.apache.ignite.internal.tx.test.ItTransactionTestUtils.withTxVoid;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.util.function.Predicate;
+import java.util.stream.IntStream;
+import org.apache.ignite.InitParametersBuilder;
+import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
+import org.apache.ignite.internal.TestWrappers;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.client.table.ClientTable;
+import org.apache.ignite.table.KeyValueView;
+import org.apache.ignite.table.Tuple;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Integration tests for the transactions running while the primary changes, not related to the tx recovery.
+ */
+public class ItClientDirectMappingTest extends ClusterPerTestIntegrationTest {
+ /** Table name. */
+ private static final String TABLE_NAME = "test_table";
+
+ /** Nodes bootstrap configuration pattern. */
+ private static final String NODE_BOOTSTRAP_CFG_TEMPLATE = "ignite {\n"
+ + " network: {\n"
+ + " port: {},\n"
+ + " nodeFinder: {\n"
+ + " netClusterNodes: [ {} ]\n"
+ + " }\n"
+ + " },\n"
+ + " clientConnector: { port:{} },\n"
+ + " rest.port: {},\n"
+ + " raft: { responseTimeout: 30000 },"
+ + " compute.threadPoolSize: 1,\n"
+ + " failureHandler.dumpThreadsOnFailure: false\n"
+ + "}";
+
+ @BeforeEach
+ public void setup() throws Exception {
+ String zoneSql = "create zone test_zone with partitions=5, replicas=1, storage_profiles='" + DEFAULT_AIPERSIST_PROFILE_NAME + "'";
+ String sql = "create table " + TABLE_NAME + " (key int primary key, val varchar(20)) zone TEST_ZONE";
+
+ cluster.doInSession(0, session -> {
+ executeUpdate(zoneSql, session);
+ executeUpdate(sql, session);
+ });
+ }
+
+ @Override
+ protected void customizeInitParameters(InitParametersBuilder builder) {
+ super.customizeInitParameters(builder);
+
+ builder.clusterConfiguration("ignite {"
+ + " transaction: {"
+ + " readOnlyTimeout: 30000,"
+ + " readWriteTimeout: 30000,"
+ + " txnResourceTtl: 2"
+ + " },"
+ + " replication: {"
+ + " rpcTimeout: 30000"
+ + " },"
+ + "}");
+ }
+
+ @Test
+ public void testBasicImplicit() {
+ try (IgniteClient client0 = clientConnectedToNode(0)) {
+ ClientTable table = (ClientTable) client0.tables().table(TABLE_NAME);
+ KeyValueView<Tuple, Tuple> kvView = table.keyValueView();
+
+ Tuple key = Tuple.create().set("key", 0);
+ Tuple val = Tuple.create().set("val", "test0");
+
+ kvView.put(null, key, val);
+ Tuple val0 = kvView.get(null, key);
+ assertTrue(Tuple.equals(val, val0));
+ }
+ }
+
+ @Test
+ public void testBasicExplicit() {
+ try (IgniteClient client0 = clientConnectedToNode(0)) {
+ ClientTable table = (ClientTable) client0.tables().table(TABLE_NAME);
+ KeyValueView<Tuple, Tuple> kvView = table.keyValueView();
+
+ Tuple key = Tuple.create().set("key", 0);
+ Tuple val = Tuple.create().set("val", "test0");
+
+ withTxVoid(client0.transactions(), tx -> kvView.put(tx, key, val));
+ Tuple val0 = withTx(client0.transactions(), tx -> kvView.get(tx, key));
+ assertTrue(Tuple.equals(val, val0));
+ }
+ }
+
+ private IgniteClient clientConnectedToNode(int nodeIndex) {
+ return IgniteClient.builder()
+ .addresses("localhost:" + igniteImpl(nodeIndex).clientAddress().port())
+ .build();
+ }
+
+ /**
+ * Returns node bootstrap config template.
+ *
+ * @return Node bootstrap config template.
+ */
+ @Override
+ protected String getNodeBootstrapConfigTemplate() {
+ return NODE_BOOTSTRAP_CFG_TEMPLATE;
+ }
+
+ private IgniteImpl findNode(int startRange, int endRange, Predicate<IgniteImpl> filter) {
+ return IntStream.range(startRange, endRange)
+ .mapToObj(this::node)
+ .map(TestWrappers::unwrapIgniteImpl)
+ .filter(filter::test)
+ .findFirst()
+ .get();
+ }
+
+ private IgniteImpl findNodeByName(String leaseholder) {
+ return findNode(0, initialNodes(), n -> leaseholder.equals(n.name()));
+ }
+
+ protected int initialNodes() {
+ return 2;
+ }
+}
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java
index fd58888024c..fc39ffebd99 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItAbstractThinClientTest.java
@@ -72,6 +72,8 @@ public abstract class ItAbstractThinClientTest extends BaseIgniteAbstractTest {
protected static final String COLUMN_VAL = "val";
+ protected static final int PARTITIONS = 10;
+
private final Map<String, String> nodesBootstrapCfg = new LinkedHashMap<>();
private final List<Ignite> startedNodes = new ArrayList<>();
@@ -132,7 +134,7 @@ public abstract class ItAbstractThinClientTest extends BaseIgniteAbstractTest {
IgniteSql sql = startedNodes.get(0).sql();
- sql.execute(null, "CREATE ZONE TEST_ZONE WITH REPLICAS=1, PARTITIONS=10, STORAGE_PROFILES='"
+ sql.execute(null, "CREATE ZONE TEST_ZONE WITH REPLICAS=1, PARTITIONS=" + PARTITIONS + ", STORAGE_PROFILES='"
+ DEFAULT_STORAGE_PROFILE + "'");
sql.execute(null, "CREATE TABLE " + TABLE_NAME + "("
+ COLUMN_KEY + " INT PRIMARY KEY, " + COLUMN_VAL + " VARCHAR) ZONE TEST_ZONE");
diff --git a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientTransactionsTest.java b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientTransactionsTest.java
index ff4daf0b103..34536f57a85 100644
--- a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientTransactionsTest.java
+++ b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/runner/app/client/ItThinClientTransactionsTest.java
@@ -17,34 +17,56 @@
package org.apache.ignite.internal.runner.app.client;
+import static java.util.Comparator.comparing;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Objects;
+import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import org.apache.ignite.Ignite;
import org.apache.ignite.client.IgniteClient;
+import org.apache.ignite.internal.TestWrappers;
+import org.apache.ignite.internal.app.IgniteImpl;
+import org.apache.ignite.internal.client.table.ClientTable;
+import org.apache.ignite.internal.client.tx.ClientLazyTransaction;
+import org.apache.ignite.internal.client.tx.ClientTransaction;
+import org.apache.ignite.internal.table.partition.HashPartition;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.internal.tx.TxState;
import org.apache.ignite.lang.ErrorGroups;
+import org.apache.ignite.lang.ErrorGroups.Transactions;
import org.apache.ignite.lang.IgniteException;
+import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.table.KeyValueView;
import org.apache.ignite.table.RecordView;
import org.apache.ignite.table.Table;
import org.apache.ignite.table.Tuple;
import org.apache.ignite.table.mapper.Mapper;
+import org.apache.ignite.table.partition.Partition;
import org.apache.ignite.tx.Transaction;
import org.apache.ignite.tx.TransactionException;
import org.apache.ignite.tx.TransactionOptions;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.Mockito;
/**
* Thin client transactions integration test.
@@ -350,6 +372,225 @@ public class ItThinClientTransactionsTest extends ItAbstractThinClientTest {
tx.rollback();
}
+ @Test
+ void testTxWithTimeout() throws InterruptedException {
+ KeyValueView<Tuple, Tuple> kvView = table().keyValueView();
+
+ Transaction tx = client().transactions().begin(new TransactionOptions().timeoutMillis(450));
+
+ // Load partition map to ensure all entries are directly mapped.
+ Map<Partition, ClusterNode> map = table().partitionManager().primaryReplicasAsync().join();
+
+ assertEquals(PARTITIONS, map.size());
+
+ int k = 111; // Avoid intersection with previous tests.
+
+ Map<Tuple, Tuple> txMap = new HashMap<>();
+
+ Tuple k1 = key(k);
+ Partition p1 = table().partitionManager().partitionAsync(k1).join();
+ Tuple v1 = val(String.valueOf(k));
+ kvView.put(tx, k1, v1);
+ txMap.put(k1, v1);
+
+ ClientTransaction tx0 = ClientLazyTransaction.get(tx).startedTx();
+
+ UUID txId = tx0.txId();
+
+ int coordIdx = -1;
+
+ for (int i = 0; i < 2; i++) {
+ Ignite server = server(i);
+ if (server.name().equals(map.get(p1).name())) {
+ coordIdx = i;
+ break;
+ }
+ }
+
+ assertTrue(coordIdx != -1);
+
+ IgniteImpl coord = TestWrappers.unwrapIgniteImpl(server(coordIdx));
+ assertNotNull(coord.txManager().stateMeta(txId), "Transaction expected to be collocated with enlistment");
+
+ IgniteImpl other = TestWrappers.unwrapIgniteImpl(server(1 - coordIdx));
+
+ do {
+ k++;
+ Tuple kt = key(k);
+ Tuple vt = val(String.valueOf(k));
+ kvView.put(tx, kt, vt);
+ txMap.put(kt, vt);
+
+ // Stop then a tx enlisted on both nodes (in direct mapping mode).
+ } while (other.txManager().stateMeta(txId) == null);
+
+ assertEquals(TxState.PENDING, coord.txManager().stateMeta(txId).txState());
+ assertEquals(TxState.PENDING, other.txManager().stateMeta(txId).txState());
+
+ assertTrue(IgniteTestUtils.waitForCondition(() -> coord.txManager().stateMeta(txId).txState() == TxState.ABORTED, 3_000),
+ "Tx is not timed out: " + coord.txManager().stateMeta(txId));
+
+ assertEquals(TxState.PENDING, other.txManager().stateMeta(txId).txState());
+
+ // Enlist after timeout should fail.
+ while (true) {
+ Tuple t = key(k);
+ Tuple v = val(String.valueOf(k));
+ try {
+ kvView.put(tx, t, v);
+
+ Thread.sleep(50);
+ } catch (TransactionException e) {
+ assertEquals(Transactions.TX_ALREADY_FINISHED_WITH_TIMEOUT_ERR, e.code());
+ break;
+ }
+ }
+
+ // A separate transaction should not wait for locks.
+ Transaction tx2 = client().transactions().begin();
+ Map<Tuple, Tuple> map2 = kvView.getAll(tx2, txMap.keySet()); // Triggers recovery.
+ assertEquals(0, map2.size());
+
+ assertEquals(TxState.ABORTED, coord.txManager().stateMeta(txId).txState());
+ assertEquals(TxState.ABORTED, other.txManager().stateMeta(txId).txState());
+
+ tx2.commit();
+ }
+
+ private List<Tuple> generateKeysForNode(int start, int count, Map<Partition, ClusterNode> map, ClusterNode clusterNode) {
+ List<Tuple> keys = new ArrayList<>();
+
+ int k = start;
+ while (keys.size() != count) {
+ k++;
+ Tuple t = key(k);
+
+ Partition part = table().partitionManager().partitionAsync(t).join();
+ ClusterNode node = map.get(part);
+
+ if (node.name().equals(clusterNode.name())) {
+ keys.add(t);
+ }
+ }
+
+ return keys;
+ }
+
+ @Test
+ void testAssignmentLoadedDuringTransaction() {
+ // Wait for assignments.
+ table().partitionManager().primaryReplicasAsync().join();
+
+ ClientTable table = (ClientTable) table();
+
+ ClientTable spyTable = Mockito.spy(table);
+
+ Map<Partition, ClusterNode> map = table().partitionManager().primaryReplicasAsync().join();
+
+ List<String> origPartMap = map.entrySet().stream().sorted(comparing(e -> {
+ HashPartition part = (HashPartition) e.getKey();
+ return part.partitionId();
+ })).map(e -> e.getValue().name()).collect(Collectors.toList());
+
+ List<String> emptyPartMap = new ArrayList<>(Collections.nCopies(10, null));
+
+ AtomicReference<List<String>> holder = new AtomicReference<>(emptyPartMap);
+
+ Mockito.doAnswer(inv -> CompletableFuture.completedFuture(holder.get())).when(spyTable).getPartitionAssignment();
+
+ ClientLazyTransaction tx0 = (ClientLazyTransaction) client().transactions().begin();
+
+ IgniteImpl server0 = TestWrappers.unwrapIgniteImpl(server(0));
+
+ List<Tuple> tuples0 = generateKeysForNode(200, 2, map, server0.clusterService().topologyService().localMember());
+
+ Tuple key = tuples0.get(0);
+ Tuple val = val(key.intValue(0) + "");
+ spyTable.keyValueView().put(tx0, key, val);
+
+ ClientTransaction tx = tx0.startedTx();
+
+ assertFalse(tx.hasCommitPartition(), "Expected proxy mode");
+
+ // Next enlistment uses loaded partition map.
+ holder.set(origPartMap);
+
+ Tuple key2 = tuples0.get(1);
+ Tuple val2 = val(key2.intValue(0) + "");
+ spyTable.keyValueView().put(tx0, key2, val2);
+
+ assertEquals(0, tx.enlistedCount(), "Expecting proxy mode");
+
+ tx.commit();
+ }
+
+ @Test
+ void testMixedMappingScenario1() {
+ Map<Partition, ClusterNode> map = table().partitionManager().primaryReplicasAsync().join();
+
+ ClientTable table = (ClientTable) table();
+
+ IgniteImpl server0 = TestWrappers.unwrapIgniteImpl(server(0));
+ IgniteImpl server1 = TestWrappers.unwrapIgniteImpl(server(1));
+
+ List<Tuple> tuples0 = generateKeysForNode(300, 1, map, server0.clusterService().topologyService().localMember());
+ List<Tuple> tuples1 = generateKeysForNode(310, 1, map, server1.clusterService().topologyService().localMember());
+
+ Map<Tuple, Tuple> data = new HashMap<>();
+
+ data.put(tuples0.get(0), val(tuples0.get(0).intValue(0) + ""));
+ data.put(tuples1.get(0), val(tuples1.get(0).intValue(0) + ""));
+
+ ClientLazyTransaction tx0 = (ClientLazyTransaction) client().transactions().begin();
+
+ table.keyValueView().putAll(tx0, data);
+
+ for (Entry<Tuple, Tuple> entry : data.entrySet()) {
+ table.keyValueView().put(tx0, entry.getKey(), entry.getValue());
+ }
+
+ tx0.commit();
+
+ for (Entry<Tuple, Tuple> entry : data.entrySet()) {
+ table.keyValueView().put(null, entry.getKey(), entry.getValue());
+ }
+ }
+
+ @Test
+ void testMixedMappingScenario2() {
+ Map<Partition, ClusterNode> map = table().partitionManager().primaryReplicasAsync().join();
+
+ ClientTable table = (ClientTable) table();
+
+ IgniteImpl server0 = TestWrappers.unwrapIgniteImpl(server(0));
+ IgniteImpl server1 = TestWrappers.unwrapIgniteImpl(server(1));
+
+ List<Tuple> tuples0 = generateKeysForNode(400, 1, map, server0.clusterService().topologyService().localMember());
+ List<Tuple> tuples1 = generateKeysForNode(410, 1, map, server1.clusterService().topologyService().localMember());
+
+ Map<Tuple, Tuple> data = new HashMap<>();
+
+ ClientLazyTransaction tx0 = (ClientLazyTransaction) client().transactions().begin();
+
+ Tuple k1 = tuples0.get(0);
+ Tuple v1 = val(tuples0.get(0).intValue(0) + "");
+ data.put(k1, v1);
+ table.keyValueView().put(tx0, k1, v1);
+
+ Tuple k2 = tuples1.get(0);
+ Tuple v2 = val(tuples1.get(0).intValue(0) + "");
+ data.put(k2, v2);
+ table.keyValueView().put(tx0, k2, v2);
+
+ table.keyValueView().putAll(tx0, data);
+
+ tx0.commit();
+
+ for (Entry<Tuple, Tuple> entry : data.entrySet()) {
+ table.keyValueView().put(null, entry.getKey(), entry.getValue());
+ }
+ }
+
private KeyValueView<Integer, String> kvView() {
return table().keyValueView(Mapper.of(Integer.class), Mapper.of(String.class));
}
@@ -358,19 +599,19 @@ public class ItThinClientTransactionsTest extends ItAbstractThinClientTest {
return client().tables().tables().get(0);
}
- private Tuple val(String v) {
+ private static Tuple val(String v) {
return Tuple.create().set(COLUMN_VAL, v);
}
- private Tuple key(Integer k) {
+ private static Tuple key(Integer k) {
return Tuple.create().set(COLUMN_KEY, k);
}
- private Tuple kv(Integer k, String v) {
+ private static Tuple kv(Integer k, String v) {
return Tuple.create().set(COLUMN_KEY, k).set(COLUMN_VAL, v);
}
- private Rec rec(int key, String val) {
+ private static Rec rec(int key, String val) {
var r = new Rec();
r.key = key;
@@ -380,8 +621,8 @@ public class ItThinClientTransactionsTest extends ItAbstractThinClientTest {
}
private static class Rec {
- public int key;
- public String val;
+ int key;
+ String val;
@Override
public boolean equals(Object o) {
diff --git a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionContextImpl.java b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionContextImpl.java
index 0997c1682eb..0c5859272e2 100644
--- a/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionContextImpl.java
+++ b/modules/sql-engine/src/main/java/org/apache/ignite/internal/sql/engine/tx/QueryTransactionContextImpl.java
@@ -72,7 +72,7 @@ public class QueryTransactionContextImpl implements QueryTransactionContext {
}
// Adding inflights only for read-only transactions. See TransactionInflights.ReadOnlyTxContext for details.
- if (transaction.isReadOnly() && !txTracker.register(transaction.id(), transaction.isReadOnly())) {
+ if (transaction.isReadOnly() && !txTracker.register(transaction.id(), true)) {
throw new TransactionException(TX_ALREADY_FINISHED_ERR, format("Transaction is already finished [tx={}]", transaction));
}
diff --git a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoOpTransaction.java b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoOpTransaction.java
index 390186f8caf..52b893483ee 100644
--- a/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoOpTransaction.java
+++ b/modules/sql-engine/src/test/java/org/apache/ignite/internal/sql/engine/framework/NoOpTransaction.java
@@ -193,11 +193,6 @@ public final class NoOpTransaction implements InternalTransaction {
return 10_000;
}
- @Override
- public long getTimeoutOrDefault(long defaultTimeout) {
- return 10_000;
- }
-
@Override
public void enlist(
ReplicationGroupId replicationGroupId,
diff --git a/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java b/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
index 0a896c025df..114ebd93ecd 100644
--- a/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
+++ b/modules/table/src/main/java/org/apache/ignite/internal/table/InternalTable.java
@@ -546,4 +546,12 @@ public interface InternalTable extends ManuallyCloseable {
* @return Streamer receiver runner.
*/
StreamerReceiverRunner streamerReceiverRunner();
+
+ /**
+ * Gets a replication group id for this table.
+ *
+ * @param partId Partition id.
+ * @return The id.
+ */
+ ReplicationGroupId targetReplicationGroupId(int partId);
}
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 f32fe3158ad..866acbb244b 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
@@ -2341,6 +2341,9 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
* @param full {@code True} if this is a full transaction.
* @param txCoordinatorId Transaction coordinator id.
* @param catalogVersion Validated catalog version associated with given operation.
+ * @param leaseStartTime Lease start time.
+ * @param skipDelayedAck {@code True} to skip delayed ack optimization.
+ *
* @return A local update ready future, possibly having a nested replication future as a result for delayed ack purpose.
*/
private CompletableFuture<CommandApplicationResult> applyUpdateCommand(
@@ -2352,6 +2355,7 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
boolean full,
UUID txCoordinatorId,
int catalogVersion,
+ boolean skipDelayedAck,
long leaseStartTime
) {
UpdateCommand cmd = updateCommand(
@@ -2368,24 +2372,42 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
);
if (!cmd.full()) {
- if (!SKIP_UPDATES) {
- // We don't need to take the partition snapshots read lock, see #INTERNAL_DOC_PLACEHOLDER why.
- storageUpdateHandler.handleUpdate(
- cmd.txId(),
- cmd.rowUuid(),
- cmd.commitPartitionId().asReplicationGroupId(),
- cmd.rowToUpdate(),
- true,
- null,
- null,
- null,
- indexIdsAtRwTxBeginTs(txId)
- );
- }
+ if (skipDelayedAck) {
+ if (!SKIP_UPDATES) {
+ storageUpdateHandler.handleUpdate(
+ cmd.txId(),
+ cmd.rowUuid(),
+ cmd.commitPartitionId().asReplicationGroupId(),
+ cmd.rowToUpdate(),
+ true,
+ null,
+ null,
+ null,
+ indexIdsAtRwTxBeginTs(txId)
+ );
+ }
- CompletableFuture<UUID> repFut = applyCmdWithExceptionHandling(cmd).thenApply(res -> cmd.txId());
+ return applyCmdWithExceptionHandling(cmd).thenApply(res -> null);
+ } else {
+ if (!SKIP_UPDATES) {
+ // We don't need to take the partition snapshots read lock, see #INTERNAL_DOC_PLACEHOLDER why.
+ storageUpdateHandler.handleUpdate(
+ cmd.txId(),
+ cmd.rowUuid(),
+ cmd.commitPartitionId().asReplicationGroupId(),
+ cmd.rowToUpdate(),
+ true,
+ null,
+ null,
+ null,
+ indexIdsAtRwTxBeginTs(txId)
+ );
+ }
- return completedFuture(new CommandApplicationResult(null, repFut));
+ CompletableFuture<UUID> repFut = applyCmdWithExceptionHandling(cmd).thenApply(res -> cmd.txId());
+
+ return completedFuture(new CommandApplicationResult(null, repFut));
+ }
} else {
return applyCmdWithExceptionHandling(cmd).thenCompose(res -> {
UpdateCommandResult updateCommandResult = (UpdateCommandResult) res;
@@ -2450,6 +2472,7 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
request.full(),
request.coordinatorId(),
catalogVersion,
+ request.skipDelayedAck(),
leaseStartTime
);
}
@@ -2846,6 +2869,7 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
request.full(),
request.coordinatorId(),
catalogVersion,
+ request.skipDelayedAck(),
leaseStartTime
)
)
@@ -2871,6 +2895,7 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
request.full(),
request.coordinatorId(),
catalogVersion,
+ request.skipDelayedAck(),
leaseStartTime
)
)
@@ -3110,6 +3135,7 @@ public class PartitionReplicaListener implements ReplicaListener, ReplicaTablePr
request.full(),
request.coordinatorId(),
catalogVersion,
+ request.skipDelayedAck(),
leaseStartTime
)
)
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 651d50b98db..5dd3dffe988 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
@@ -395,7 +395,7 @@ public class InternalTableImpl implements InternalTable {
return postEnlist(fut, false, actualTx, actualTx.implicit()).handle((r, e) -> {
if (e != null) {
if (actualTx.implicit()) {
- long timeout = getTimeoutOrDefault(actualTx);
+ long timeout = actualTx.getTimeout();
long ts = (txStartTs == null) ? actualTx.startTimestamp().getPhysical() : txStartTs;
@@ -515,7 +515,7 @@ public class InternalTableImpl implements InternalTable {
return postEnlist(fut, actualTx.implicit() && !singlePart, actualTx, full).handle((r, e) -> {
if (e != null) {
if (actualTx.implicit()) {
- long timeout = getTimeoutOrDefault(actualTx);
+ long timeout = actualTx.getTimeout();
long ts = (txStartTs == null) ? actualTx.startTimestamp().getPhysical() : txStartTs;
@@ -688,7 +688,7 @@ public class InternalTableImpl implements InternalTable {
});
} else {
if (write) { // Track only write requests from explicit transactions.
- if (!transactionInflights.addInflight(tx.id(), false)) {
+ if (!tx.remote() && !transactionInflights.addInflight(tx.id(), false)) {
int code = TX_ALREADY_FINISHED_ERR;
if (tx.isRolledBackWithTimeoutExceeded()) {
code = TX_ALREADY_FINISHED_WITH_TIMEOUT_ERR;
@@ -708,7 +708,7 @@ public class InternalTableImpl implements InternalTable {
assert noWriteChecker != null;
// Remove inflight if no replication was scheduled, otherwise inflight will be removed by delayed response.
- if (noWriteChecker.test(res, request)) {
+ if (!tx.remote() && noWriteChecker.test(res, request)) {
transactionInflights.removeInflight(tx.id());
}
@@ -716,7 +716,9 @@ public class InternalTableImpl implements InternalTable {
}).handle((r, e) -> {
if (e != null) {
if (retryOnLockConflict > 0 && matchAny(unwrapCause(e), ACQUIRE_LOCK_ERR)) {
- transactionInflights.removeInflight(tx.id()); // Will be retried.
+ if (!tx.remote()) {
+ transactionInflights.removeInflight(tx.id()); // Will be retried.
+ }
return trackingInvoke(
tx,
@@ -774,7 +776,7 @@ public class InternalTableImpl implements InternalTable {
assert !(autoCommit && full) : "Invalid combination of flags";
return fut.handle((BiFunction<T, Throwable, CompletableFuture<T>>) (r, e) -> {
- if (full) {
+ if (full || tx0.remote()) {
return e != null ? failedFuture(e) : completedFuture(r);
}
@@ -962,6 +964,7 @@ public class InternalTableImpl implements InternalTable {
.timestamp(txo.startTimestamp())
.full(false)
.coordinatorId(txo.coordinatorId())
+ .skipDelayedAck(txo.remote())
.build(),
(res, req) -> false
);
@@ -1172,6 +1175,7 @@ public class InternalTableImpl implements InternalTable {
.timestamp(txo.startTimestamp())
.full(txo.implicit())
.coordinatorId(txo.coordinatorId())
+ .skipDelayedAck(txo.remote())
.build(),
(res, req) -> false
);
@@ -1227,7 +1231,7 @@ public class InternalTableImpl implements InternalTable {
// Will be finished in one RTT.
return postEnlist(fut, false, tx, true).handle((r, e) -> {
if (e != null) {
- long timeout = getTimeoutOrDefault(tx);
+ long timeout = tx.getTimeout();
long ts = (txStartTs == null) ? tx.startTimestamp().getPhysical() : txStartTs;
@@ -1242,10 +1246,6 @@ public class InternalTableImpl implements InternalTable {
}).thenCompose(identity());
}
- private long getTimeoutOrDefault(InternalTransaction tx) {
- return tx.getTimeoutOrDefault(getDefaultTimeout(tx));
- }
-
private long getDefaultTimeout(InternalTransaction tx) {
return tx.isReadOnly() ? defaultReadTxTimeout.get() : defaultRwTxTimeout.get();
}
@@ -1270,6 +1270,7 @@ public class InternalTableImpl implements InternalTable {
.timestamp(txo.startTimestamp())
.full(txo.implicit())
.coordinatorId(txo.coordinatorId())
+ .skipDelayedAck(txo.remote())
.build(),
(res, req) -> false
);
@@ -1293,6 +1294,7 @@ public class InternalTableImpl implements InternalTable {
.timestamp(txo.startTimestamp())
.full(txo.implicit())
.coordinatorId(txo.coordinatorId())
+ .skipDelayedAck(txo.remote())
.build(),
(res, req) -> !res
);
@@ -1351,6 +1353,7 @@ public class InternalTableImpl implements InternalTable {
.timestamp(tx.startTimestamp())
.full(full)
.coordinatorId(tx.coordinatorId())
+ .skipDelayedAck(tx.remote())
.build();
}
@@ -1372,6 +1375,7 @@ public class InternalTableImpl implements InternalTable {
.timestamp(txo.startTimestamp())
.full(txo.implicit())
.coordinatorId(txo.coordinatorId())
+ .skipDelayedAck(txo.remote())
.build(),
(res, req) -> !res
);
@@ -1399,6 +1403,7 @@ public class InternalTableImpl implements InternalTable {
.timestamp(txo.startTimestamp())
.full(txo.implicit())
.coordinatorId(txo.coordinatorId())
+ .skipDelayedAck(txo.remote())
.build(),
(res, req) -> !res
);
@@ -1424,6 +1429,7 @@ public class InternalTableImpl implements InternalTable {
.timestamp(txo.startTimestamp())
.full(txo.implicit())
.coordinatorId(txo.coordinatorId())
+ .skipDelayedAck(txo.remote())
.build(),
(res, req) -> res == null
);
@@ -1447,6 +1453,7 @@ public class InternalTableImpl implements InternalTable {
.timestamp(txo.startTimestamp())
.full(txo.implicit())
.coordinatorId(txo.coordinatorId())
+ .skipDelayedAck(txo.remote())
.build(),
(res, req) -> !res
);
@@ -1470,6 +1477,7 @@ public class InternalTableImpl implements InternalTable {
.timestamp(txo.startTimestamp())
.full(txo.implicit())
.coordinatorId(txo.coordinatorId())
+ .skipDelayedAck(txo.remote())
.build(),
(res, req) -> !res
);
@@ -1495,6 +1503,7 @@ public class InternalTableImpl implements InternalTable {
.timestamp(txo.startTimestamp())
.full(txo.implicit())
.coordinatorId(txo.coordinatorId())
+ .skipDelayedAck(txo.remote())
.build(),
(res, req) -> res == null
);
@@ -2228,7 +2237,8 @@ public class InternalTableImpl implements InternalTable {
.thenApply(v -> Arrays.stream(invokeFutures).mapToLong(f -> (Long) f.join()).sum());
}
- private ReplicationGroupId targetReplicationGroupId(int partId) {
+ @Override
+ public final ReplicationGroupId targetReplicationGroupId(int partId) {
if (enabledColocation()) {
return new ZonePartitionId(zoneId, partId);
} else {
diff --git a/modules/transactions/src/jmh/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryBenchmark.java b/modules/transactions/src/jmh/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryBenchmark.java
index dccfea66950..a7eb511246c 100644
--- a/modules/transactions/src/jmh/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryBenchmark.java
+++ b/modules/transactions/src/jmh/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryBenchmark.java
@@ -185,11 +185,6 @@ public class TransactionExpirationRegistryBenchmark {
return 0;
}
- @Override
- public long getTimeoutOrDefault(long defaultTimeout) {
- return 0;
- }
-
@Override
public CompletableFuture<Void> kill() {
return null;
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/InternalTransaction.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/InternalTransaction.java
index 4b919e097ab..b20c99a39d8 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/InternalTransaction.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/InternalTransaction.java
@@ -112,6 +112,15 @@ public interface InternalTransaction extends Transaction {
*/
boolean implicit();
+ /**
+ * Gets the transaction remote flag.
+ *
+ * @return True if the transaction is remotely coordinated, false otherwise.
+ */
+ default boolean remote() {
+ return false;
+ }
+
/**
* Finishes a read-only transaction with a specific execution timestamp.
*
@@ -139,13 +148,6 @@ public interface InternalTransaction extends Transaction {
*/
long getTimeout();
- /**
- * Returns the transaction timeout in millis or the default timeout if the transaction timeout is set to 0.
- *
- * @return The transaction timeout.
- */
- long getTimeoutOrDefault(long defaultTimeout);
-
/**
* Kills this transaction.
*
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java
index e03ebea41f3..94936c08cf4 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/TxManager.java
@@ -26,6 +26,7 @@ import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.hlc.HybridTimestampTracker;
import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.TablePartitionId;
import org.apache.ignite.internal.tx.impl.EnlistedPartitionGroup;
import org.jetbrains.annotations.Nullable;
import org.jetbrains.annotations.TestOnly;
@@ -102,6 +103,18 @@ public interface TxManager extends IgniteComponent {
*/
InternalTransaction beginExplicit(HybridTimestampTracker timestampTracker, boolean readOnly, InternalTxOptions txOptions);
+ /**
+ * Begins a remote transaction.
+ *
+ * @param txId Tx id.
+ * @param commitPartId Table partition id.
+ * @param coord Tx coordinator.
+ * @param token Enlistment token.
+ * @param timeout The timeout.
+ * @return Remote transaction.
+ */
+ InternalTransaction beginRemote(UUID txId, TablePartitionId commitPartId, UUID coord, long token, long timeout);
+
/**
* Returns a transaction state meta.
*
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PublicApiThreadingTransaction.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PublicApiThreadingTransaction.java
index 03e7e7eca73..2da41d32eb4 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PublicApiThreadingTransaction.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/PublicApiThreadingTransaction.java
@@ -149,11 +149,6 @@ public class PublicApiThreadingTransaction implements InternalTransaction, Wrapp
return transaction.getTimeout();
}
- @Override
- public long getTimeoutOrDefault(long defaultTimeout) {
- return transaction.getTimeoutOrDefault(defaultTimeout);
- }
-
@Override
public <T> T unwrap(Class<T> classToUnwrap) {
return classToUnwrap.cast(transaction);
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyTransactionImpl.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyTransactionImpl.java
index 342da132cdc..e3e6f2cc551 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyTransactionImpl.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadOnlyTransactionImpl.java
@@ -51,6 +51,7 @@ public class ReadOnlyTransactionImpl extends IgniteAbstractTransactionImpl {
* @param id The id.
* @param txCoordinatorId Transaction coordinator inconsistent ID.
* @param implicit True for an implicit transaction, false for an ordinary one.
+ * @param timeout The timeout.
* @param readTimestamp The read timestamp.
*/
ReadOnlyTransactionImpl(
@@ -158,11 +159,6 @@ public class ReadOnlyTransactionImpl extends IgniteAbstractTransactionImpl {
return finishGuard.get();
}
- @Override
- public long getTimeoutOrDefault(long defaultTimeout) {
- return timeout == 0 ? defaultTimeout : timeout;
- }
-
@Override
public CompletableFuture<Void> kill() {
return finish(false, readTimestamp, false, false);
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadWriteTransactionImpl.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadWriteTransactionImpl.java
index 2a75ab02239..7e17069d1a3 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadWriteTransactionImpl.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/ReadWriteTransactionImpl.java
@@ -74,6 +74,7 @@ public class ReadWriteTransactionImpl extends IgniteAbstractTransactionImpl {
* @param id The id.
* @param txCoordinatorId Transaction coordinator inconsistent ID.
* @param implicit True for an implicit transaction, false for an ordinary one.
+ * @param timeout The timeout.
*/
public ReadWriteTransactionImpl(
TxManager txManager,
@@ -206,6 +207,7 @@ public class ReadWriteTransactionImpl extends IgniteAbstractTransactionImpl {
* @param executionTimestamp The timestamp is the time when the transaction is applied to the remote node.
* @param full Full state transaction marker.
* @param isComplete The flag is true if the transaction is completed through the public API, false for {@link this#kill()} invocation.
+ * @param timeoutExceeded {@code True} if rollback reason is the timeout.
* @return The future.
*/
private CompletableFuture<Void> finishInternal(
@@ -274,11 +276,6 @@ public class ReadWriteTransactionImpl extends IgniteAbstractTransactionImpl {
return finishFuture != null;
}
- @Override
- public long getTimeoutOrDefault(long defaultTimeout) {
- return timeout == USE_CONFIGURED_TIMEOUT_DEFAULT ? defaultTimeout : timeout;
- }
-
/** {@inheritDoc} */
@Override
public boolean isReadOnly() {
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/RemoteReadWriteTransaction.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/RemoteReadWriteTransaction.java
new file mode 100644
index 00000000000..b6fc52ae793
--- /dev/null
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/RemoteReadWriteTransaction.java
@@ -0,0 +1,167 @@
+/*
+ * 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.CompletableFuture;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.apache.ignite.internal.tostring.IgniteToStringExclude;
+import org.apache.ignite.internal.tostring.S;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.tx.PendingTxPartitionEnlistment;
+import org.apache.ignite.internal.tx.TransactionIds;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.tx.TransactionException;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Remote read-write transaction which is coordinated from elsewhere.
+ * Used for a direct mapping request from a client.
+ */
+public abstract class RemoteReadWriteTransaction implements InternalTransaction {
+ private static final String EXCEPTION_MSG = "Remote transaction should never be finished directly";
+
+ private final UUID txId;
+ private final TablePartitionId commitGroupId;
+ private final long timeout;
+ private final UUID coord;
+ @IgniteToStringExclude
+ private @Nullable PendingTxPartitionEnlistment enlistment;
+
+ /**
+ * The constructor.
+ *
+ * @param txId Tx id.
+ * @param commitGroupId Commit group id.
+ * @param coord The coordinator id.
+ * @param token Enlistment token.
+ * @param localNode Local node.
+ * @param timeout The timeout.
+ */
+ RemoteReadWriteTransaction(UUID txId, TablePartitionId commitGroupId, UUID coord, long token, ClusterNode localNode,
+ long timeout) {
+ this.txId = txId;
+ this.commitGroupId = commitGroupId;
+ this.coord = coord;
+ this.timeout = timeout;
+ this.enlistment = token == 0 ? null : new PendingTxPartitionEnlistment(localNode.name(), token);
+ }
+
+ @Override
+ public void commit() throws TransactionException {
+ throw new AssertionError(EXCEPTION_MSG);
+ }
+
+ @Override
+ public CompletableFuture<Void> commitAsync() {
+ throw new AssertionError(EXCEPTION_MSG);
+ }
+
+ @Override
+ public void rollback() throws TransactionException {
+ throw new AssertionError(EXCEPTION_MSG);
+ }
+
+ @Override
+ public CompletableFuture<Void> rollbackAsync() {
+ throw new AssertionError(EXCEPTION_MSG);
+ }
+
+ @Override
+ public boolean isReadOnly() {
+ return false;
+ }
+
+ @Override
+ public UUID id() {
+ return txId;
+ }
+
+ @Override
+ public PendingTxPartitionEnlistment enlistedPartition(ReplicationGroupId replicationGroupId) {
+ return enlistment;
+ }
+
+ @Override
+ public boolean assignCommitPartition(ReplicationGroupId replicationGroupId) {
+ return false;
+ }
+
+ @Override
+ public TablePartitionId commitPartition() {
+ return commitGroupId;
+ }
+
+ @Override
+ public void enlist(ReplicationGroupId replicationGroupId, int tableId, String primaryNodeConsistentId, long consistencyToken) {
+ this.enlistment = new PendingTxPartitionEnlistment(primaryNodeConsistentId, consistencyToken, tableId);
+ }
+
+ @Override
+ public @Nullable HybridTimestamp readTimestamp() {
+ return null;
+ }
+
+ @Override
+ public HybridTimestamp startTimestamp() {
+ return TransactionIds.beginTimestamp(txId);
+ }
+
+ @Override
+ public UUID coordinatorId() {
+ return coord;
+ }
+
+ @Override
+ public boolean implicit() {
+ return false;
+ }
+
+ @Override
+ public boolean remote() {
+ return true;
+ }
+
+ @Override
+ public CompletableFuture<Void> finish(boolean commit, @Nullable HybridTimestamp executionTimestamp, boolean full,
+ boolean timeoutExceeded) {
+ return null;
+ }
+
+ @Override
+ public boolean isFinishingOrFinished() {
+ return false;
+ }
+
+ @Override
+ public long getTimeout() {
+ return timeout;
+ }
+
+ @Override
+ public CompletableFuture<Void> kill() {
+ return null;
+ }
+
+ @Override
+ public String toString() {
+ return S.toString(RemoteReadWriteTransaction.class, this);
+ }
+}
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistry.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistry.java
index 4b7a3a8a9ed..15bece929d1 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistry.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistry.java
@@ -20,16 +20,16 @@ package org.apache.ignite.internal.tx.impl;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
-import java.util.Map;
import java.util.NavigableMap;
import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.tx.TransactionIds;
class TransactionExpirationRegistry {
private static final IgniteLogger LOG = Loggers.forClass(TransactionExpirationRegistry.class);
@@ -40,14 +40,33 @@ class TransactionExpirationRegistry {
*/
private final NavigableMap<Long, Object> txsByExpirationTime = new ConcurrentSkipListMap<>();
- /** Map from registered transaction to its expiration timestamp. */
- private final Map<InternalTransaction, Long> expirationTimeByTx = new ConcurrentHashMap<>();
-
private final ReadWriteLock watermarkLock = new ReentrantReadWriteLock();
/** Watermark at which expiration has already happened (millis since Unix epoch). */
private volatile long watermark = Long.MIN_VALUE;
+ private static long physicalExpirationTimeMillis(HybridTimestamp beginTimestamp, long effectiveTimeoutMillis) {
+ return sumWithSaturation(beginTimestamp.getPhysical(), effectiveTimeoutMillis);
+ }
+
+ private static long sumWithSaturation(long a, long b) {
+ assert a >= 0 : a;
+ assert b >= 0 : b;
+
+ long sum = a + b;
+
+ if (sum < 0) {
+ // Overflow.
+ return Long.MAX_VALUE;
+ } else {
+ return sum;
+ }
+ }
+
+ void register(InternalTransaction tx) {
+ register(tx, physicalExpirationTimeMillis(TransactionIds.beginTimestamp(tx.id()), tx.getTimeout()));
+ }
+
void register(InternalTransaction tx, long txExpirationTime) {
if (isExpired(txExpirationTime)) {
abortTransaction(tx);
@@ -82,8 +101,6 @@ class TransactionExpirationRegistry {
return txsExpiringAtTs;
}
);
-
- expirationTimeByTx.put(tx, txExpirationTime);
} finally {
watermarkLock.readLock().unlock();
}
@@ -96,7 +113,7 @@ class TransactionExpirationRegistry {
private static void abortTransaction(InternalTransaction tx) {
tx.rollbackTimeoutExceededAsync().whenComplete((res, ex) -> {
if (ex != null) {
- LOG.error("Transaction abort due to timeout failed [txId={}]", ex, tx.id());
+ LOG.error("Transaction has aborted due to timeout [txId={}]", ex, tx.id());
}
});
}
@@ -119,13 +136,11 @@ class TransactionExpirationRegistry {
for (Object txOrSet : transactionsAndSetsToExpire) {
if (txOrSet instanceof Set) {
for (InternalTransaction tx : (Set<InternalTransaction>) txOrSet) {
- expirationTimeByTx.remove(tx);
abortTransaction(tx);
}
} else {
InternalTransaction tx = (InternalTransaction) txOrSet;
- expirationTimeByTx.remove(tx);
abortTransaction(tx);
}
}
@@ -136,20 +151,26 @@ class TransactionExpirationRegistry {
}
void unregister(InternalTransaction tx) {
- Long expirationTime = expirationTimeByTx.remove(tx);
+ unregister(tx, physicalExpirationTimeMillis(TransactionIds.beginTimestamp(tx.id()), tx.getTimeout()));
+ }
- if (expirationTime != null) {
- txsByExpirationTime.computeIfPresent(expirationTime, (k, txOrSet) -> {
- if (txOrSet instanceof Set) {
- Set<InternalTransaction> set = (Set<InternalTransaction>) txOrSet;
+ void unregister(InternalTransaction tx, long expirationTime) {
+ txsByExpirationTime.computeIfPresent(expirationTime, (k, txOrSet) -> {
+ if (txOrSet instanceof Set) {
+ Set<InternalTransaction> set = (Set<InternalTransaction>) txOrSet;
- set.remove(tx);
+ set.remove(tx);
+
+ return set.size() == 1 ? set.iterator().next() : set;
+ } else {
+ InternalTransaction tx0 = (InternalTransaction) txOrSet;
- return set.size() == 1 ? set.iterator().next() : set;
- } else {
+ if (tx0.id().equals(tx.id())) {
return null;
}
- });
- }
+
+ return tx0;
+ }
+ });
}
}
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 0a34be9ee57..22e0f267404 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
@@ -124,6 +124,9 @@ import org.jetbrains.annotations.TestOnly;
* <p>Uses 2PC for atomic commitment and 2PL for concurrency control.
*/
public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemViewProvider {
+ /** Expirarion trigger frequency. */
+ public static final long EXPIRE_FREQ_MILLIS = 1000;
+
/** The logger. */
private static final IgniteLogger LOG = Loggers.forClass(TxManagerImpl.class);
@@ -451,7 +454,7 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
boolean scheduleExpiration = !implicit;
if (scheduleExpiration) {
- transactionExpirationRegistry.register(transaction, physicalExpirationTimeMillis(beginTimestamp, timeout));
+ transactionExpirationRegistry.register(transaction);
}
return transaction;
@@ -503,7 +506,7 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
boolean scheduleExpiration = !implicit;
if (scheduleExpiration) {
- transactionExpirationRegistry.register(transaction, physicalExpirationTimeMillis(beginTimestamp, timeout));
+ transactionExpirationRegistry.register(transaction);
}
txFuture.whenComplete((unused, throwable) -> {
@@ -522,24 +525,6 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
}
}
- private static long physicalExpirationTimeMillis(HybridTimestamp beginTimestamp, long effectiveTimeoutMillis) {
- return sumWithSaturation(beginTimestamp.getPhysical(), effectiveTimeoutMillis);
- }
-
- private static long sumWithSaturation(long a, long b) {
- assert a >= 0 : a;
- assert b >= 0 : b;
-
- long sum = a + b;
-
- if (sum < 0) {
- // Overflow.
- return Long.MAX_VALUE;
- } else {
- return sum;
- }
- }
-
/**
* Current read timestamp, for calculation of read timestamp of read-only transactions.
*
@@ -892,6 +877,48 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
return startedTxs.intValue() - finishedTxs.intValue();
}
+ @Override
+ public InternalTransaction beginRemote(UUID txId, TablePartitionId commitPartId, UUID coord, long token, long timeout) {
+ assert commitPartId.tableId() > 0 && commitPartId.partitionId() >= 0 : "Illegal condition for direct mapping: " + commitPartId;
+
+ // Switch to default timeout if needed.
+ timeout = timeout == USE_CONFIGURED_TIMEOUT_DEFAULT ? txConfig.readWriteTimeout().value() : timeout;
+
+ // Adjust the timeout so local expiration happens after coordinator expiration.
+ var tx = new RemoteReadWriteTransaction(txId, commitPartId, coord, token, topologyService.localMember(),
+ timeout + clockService.maxClockSkewMillis()) {
+ boolean isTimeout = false;
+
+ @Override
+ public TxState state() {
+ TxStateMeta meta = TxManagerImpl.this.stateMeta(txId);
+
+ return meta == null ? null : meta.txState();
+ }
+
+ @Override
+ public CompletableFuture<Void> rollbackTimeoutExceededAsync() {
+ isTimeout = true;
+
+ // Directly mapped entries become abandoned on local tx timeout.
+ // Release locks to allow write intent resolution on abandoned path.
+ // Can be safely retried multiple times, because releaseAll is idempotent.
+ partitionOperationsExecutor.execute(() -> lockManager.releaseAll(txId));
+
+ return nullCompletedFuture();
+ }
+
+ @Override
+ public boolean isRolledBackWithTimeoutExceeded() {
+ return isTimeout;
+ }
+ };
+
+ transactionExpirationRegistry.register(tx);
+
+ return tx;
+ }
+
@Override
public CompletableFuture<Void> startAsync(ComponentContext componentContext) {
return inBusyLockAsync(busyLock, () -> {
@@ -924,7 +951,8 @@ public class TxManagerImpl implements TxManager, NetworkMessageHandler, SystemVi
placementDriver.listen(PrimaryReplicaEvent.PRIMARY_REPLICA_ELECTED, primaryReplicaElectedListener);
- transactionExpirationJobFuture = commonScheduler.scheduleAtFixedRate(this::expireTransactionsUpToNow, 1000, 1000, MILLISECONDS);
+ transactionExpirationJobFuture = commonScheduler.scheduleAtFixedRate(this::expireTransactionsUpToNow,
+ EXPIRE_FREQ_MILLIS, EXPIRE_FREQ_MILLIS, MILLISECONDS);
return nullCompletedFuture();
});
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryConcurrentTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryConcurrentTest.java
index dd56a1ab131..ebff79347ee 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryConcurrentTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryConcurrentTest.java
@@ -117,7 +117,7 @@ class TransactionExpirationRegistryConcurrentTest extends BaseIgniteAbstractTest
for (int i = 0; i < count; i++) {
if (i % 2 == 0) {
- transactionExpirationRegistry.unregister(txPool.get(i));
+ transactionExpirationRegistry.unregister(txPool.get(i), offset + i);
continue;
}
@@ -181,7 +181,7 @@ class TransactionExpirationRegistryConcurrentTest extends BaseIgniteAbstractTest
@Override
public UUID id() {
- return UUID.fromString(id + "");
+ return new UUID(id, 0);
}
@Override
@@ -246,11 +246,6 @@ class TransactionExpirationRegistryConcurrentTest extends BaseIgniteAbstractTest
return 0;
}
- @Override
- public long getTimeoutOrDefault(long defaultTimeout) {
- return defaultTimeout;
- }
-
@Override
public CompletableFuture<Void> kill() {
return null;
diff --git a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryTest.java b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryTest.java
index d10cd3e4297..edb4f504b06 100644
--- a/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryTest.java
+++ b/modules/transactions/src/test/java/org/apache/ignite/internal/tx/impl/TransactionExpirationRegistryTest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.tx.impl;
+import static org.apache.ignite.internal.tx.TransactionIds.transactionId;
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
import static org.mockito.Mockito.lenient;
@@ -24,6 +25,9 @@ import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import java.util.UUID;
+import org.apache.ignite.internal.TestHybridClock;
+import org.apache.ignite.internal.hlc.HybridClock;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.junit.jupiter.api.BeforeEach;
@@ -44,6 +48,13 @@ class TransactionExpirationRegistryTest extends BaseIgniteAbstractTest {
@BeforeEach
void configureMocks() {
+ HybridClock clock = new TestHybridClock(() -> 0);
+
+ UUID txId1 = transactionId(clock.now(), 0);
+ UUID txId2 = transactionId(clock.now(), 0);
+
+ lenient().when(tx1.id()).thenReturn(txId1);
+ lenient().when(tx2.id()).thenReturn(txId2);
lenient().when(tx1.rollbackTimeoutExceededAsync()).thenReturn(nullCompletedFuture());
lenient().when(tx2.rollbackTimeoutExceededAsync()).thenReturn(nullCompletedFuture());
}
@@ -148,6 +159,8 @@ class TransactionExpirationRegistryTest extends BaseIgniteAbstractTest {
void removesTransactionOnUnregister() {
registry.register(tx1, 1000);
+ lenient().when(tx1.getTimeout()).thenReturn(1000L);
+
registry.unregister(tx1);
registry.expireUpTo(2000);
@@ -160,6 +173,8 @@ class TransactionExpirationRegistryTest extends BaseIgniteAbstractTest {
void unregisterIsIdempotent() {
registry.register(tx1, 1000);
+ lenient().when(tx1.getTimeout()).thenReturn(1000L);
+
registry.unregister(tx1);
assertDoesNotThrow(() -> registry.unregister(tx1));
diff --git a/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/ItTransactionTestUtils.java b/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/ItTransactionTestUtils.java
index 94db5e60c0e..125ba5a0a7b 100644
--- a/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/ItTransactionTestUtils.java
+++ b/modules/transactions/src/testFixtures/java/org/apache/ignite/internal/tx/test/ItTransactionTestUtils.java
@@ -30,6 +30,7 @@ import java.util.HashSet;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
import java.util.function.Function;
import org.apache.ignite.Ignite;
import org.apache.ignite.internal.app.IgniteImpl;
@@ -49,6 +50,7 @@ import org.apache.ignite.internal.wrapper.Wrappers;
import org.apache.ignite.table.RecordView;
import org.apache.ignite.table.Table;
import org.apache.ignite.table.Tuple;
+import org.apache.ignite.tx.IgniteTransactions;
import org.apache.ignite.tx.Transaction;
import org.jetbrains.annotations.Nullable;
@@ -216,6 +218,31 @@ public class ItTransactionTestUtils {
return primaryReplicaFut.join();
}
+ /**
+ * Executes a closure in a transaction.
+ *
+ * @param transactions Transactions facade.
+ * @param c The closure.
+ */
+ public static void withTxVoid(IgniteTransactions transactions, Consumer<Transaction> c) {
+ Transaction tx = transactions.begin();
+ c.accept(tx);
+ tx.commit();
+ }
+
+ /**
+ * Executes a closure in a transaction.
+ *
+ * @param transactions Transactions facade.
+ * @param c The closure.
+ */
+ public static <T> T withTx(IgniteTransactions transactions, Function<Transaction, T> c) {
+ Transaction tx = transactions.begin();
+ T t = c.apply(tx);
+ tx.commit();
+ return t;
+ }
+
/**
* Unwraps {@link RecordBinaryViewImpl} from a {@link RecordView}.
*