You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by pp...@apache.org on 2025/01/20 12:06:35 UTC
(ignite-3) branch ignite-24053 created (now e11524580fc)
This is an automated email from the ASF dual-hosted git repository.
ppa pushed a change to branch ignite-24053
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
at e11524580fc IGNITE-24053 Use single obserable time tracker for multiple JDBC connections.
This branch includes the following new commits:
new e11524580fc IGNITE-24053 Use single obserable time tracker for multiple JDBC connections.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(ignite-3) 01/01: IGNITE-24053 Use single obserable time tracker for multiple JDBC connections.
Posted by pp...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ppa pushed a commit to branch ignite-24053
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
commit e11524580fc4a2bb9928bfbd456f70e652684092
Author: Pavel Pereslegin <xx...@gmail.com>
AuthorDate: Mon Jan 20 14:54:37 2025 +0300
IGNITE-24053 Use single obserable time tracker for multiple JDBC connections.
---
.../jdbc/proto/event/JdbcBatchExecuteRequest.java | 2 +-
.../proto/event/JdbcBatchPreparedStmntRequest.java | 2 +-
.../jdbc/proto/event/JdbcFinishTxResult.java | 25 ++-
...lt.java => JdbcObservableTimeAwareRequest.java} | 27 ++-
.../jdbc/proto/event/JdbcQueryExecuteRequest.java | 14 +-
.../ignite/internal/jdbc/proto/event/Response.java | 7 +-
.../handler/ClientInboundMessageHandler.java | 4 -
.../client/handler/JdbcConnectionContext.java | 140 +++++++++++++++
.../ignite/client/handler/JdbcHandlerBase.java | 5 -
.../client/handler/JdbcQueryEventHandlerImpl.java | 153 ++++------------
.../jdbc/ClientJdbcExecuteBatchRequest.java | 4 +-
.../requests/jdbc/ClientJdbcExecuteRequest.java | 5 +-
.../requests/jdbc/ClientJdbcFinishTxRequest.java | 2 +-
.../jdbc/ClientJdbcPreparedStmntBatchRequest.java | 4 +-
.../sql/ClientSqlExecuteScriptRequest.java | 1 +
.../handler/JdbcQueryEventHandlerImplTest.java | 194 +++++++++++++++++++--
.../ignite/internal/client/ReliableChannel.java | 8 +-
.../ignite/internal/client/TcpIgniteClient.java | 24 ++-
.../apache/ignite/client/fakes/FakeTxManager.java | 5 +
.../ignite/jdbc/ItJdbcMultipleConnectionsTest.java | 84 +++++++++
.../ignite/internal/jdbc/JdbcConnection.java | 38 +++-
.../apache/ignite/internal/jdbc/JdbcStatement.java | 3 +-
.../org/apache/ignite/jdbc/IgniteJdbcDriver.java | 5 +-
.../sql/engine/framework/NoOpTransaction.java | 6 +
.../ignite/internal/tx/HybridTimestampTracker.java | 24 +++
.../ignite/internal/tx/InternalTransaction.java | 7 +
.../tx/impl/IgniteAbstractTransactionImpl.java | 7 +
.../tx/impl/PublicApiThreadingTransaction.java | 5 +
28 files changed, 605 insertions(+), 200 deletions(-)
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchExecuteRequest.java b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchExecuteRequest.java
index 0994126b837..31ed5e6844a 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchExecuteRequest.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchExecuteRequest.java
@@ -28,7 +28,7 @@ import org.apache.ignite.internal.util.CollectionUtils;
/**
* JDBC batch execute request.
*/
-public class JdbcBatchExecuteRequest implements ClientMessage {
+public class JdbcBatchExecuteRequest extends JdbcObservableTimeAwareRequest implements ClientMessage {
/** Schema name. */
private String schemaName;
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchPreparedStmntRequest.java b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchPreparedStmntRequest.java
index 6391b29a17a..ddf36c2972a 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchPreparedStmntRequest.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcBatchPreparedStmntRequest.java
@@ -29,7 +29,7 @@ import org.apache.ignite.internal.util.CollectionUtils;
/**
* JDBC prepared statement query batch execute request.
*/
-public class JdbcBatchPreparedStmntRequest implements ClientMessage {
+public class JdbcBatchPreparedStmntRequest extends JdbcObservableTimeAwareRequest implements ClientMessage {
/** Schema name. */
private String schemaName;
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcFinishTxResult.java b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcFinishTxResult.java
index 5c6cbfa6f26..1a6efd904c1 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcFinishTxResult.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcFinishTxResult.java
@@ -17,15 +17,29 @@
package org.apache.ignite.internal.jdbc.proto.event;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.jetbrains.annotations.Nullable;
+
/**
* Result of commit/rollback command.
*/
public class JdbcFinishTxResult extends Response {
+ /** Observable timestamp used only on server side. */
+ @SuppressWarnings("TransientFieldInNonSerializableClass")
+ private final transient @Nullable HybridTimestamp observableTime;
+
/**
* Default constructor is used for deserialization.
*/
public JdbcFinishTxResult() {
- // No-op.
+ this.observableTime = null;
+ }
+
+ /**
+ * Constructor.
+ */
+ public JdbcFinishTxResult(@Nullable HybridTimestamp observableTime) {
+ this.observableTime = observableTime;
}
/**
@@ -36,5 +50,14 @@ public class JdbcFinishTxResult extends Response {
*/
public JdbcFinishTxResult(int status, String err) {
super(status, err);
+
+ this.observableTime = null;
+ }
+
+ /**
+ * Returns transaction observable time.
+ */
+ public @Nullable HybridTimestamp observableTime() {
+ return observableTime;
}
}
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcFinishTxResult.java b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcObservableTimeAwareRequest.java
similarity index 56%
copy from modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcFinishTxResult.java
copy to modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcObservableTimeAwareRequest.java
index 5c6cbfa6f26..fffe37a087e 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcFinishTxResult.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcObservableTimeAwareRequest.java
@@ -17,24 +17,19 @@
package org.apache.ignite.internal.jdbc.proto.event;
+import java.util.concurrent.atomic.AtomicReference;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+
/**
- * Result of commit/rollback command.
+ * An extension to JDBC request that provides the ability to update and read client observable time.
*/
-public class JdbcFinishTxResult extends Response {
- /**
- * Default constructor is used for deserialization.
- */
- public JdbcFinishTxResult() {
- // No-op.
- }
+abstract class JdbcObservableTimeAwareRequest {
+ /** Holds the latest time observed by client. */
+ @SuppressWarnings("TransientFieldInNonSerializableClass")
+ private final transient AtomicReference<HybridTimestamp> observableTimeHolder = new AtomicReference<>();
- /**
- * Constructor.
- *
- * @param status Status code.
- * @param err Error message.
- */
- public JdbcFinishTxResult(int status, String err) {
- super(status, err);
+ /** Returns the holder containing the latest time observed by the client. */
+ public AtomicReference<HybridTimestamp> observableTimeHolder() {
+ return observableTimeHolder;
}
}
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcQueryExecuteRequest.java b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcQueryExecuteRequest.java
index cd48d51d1f8..f37d3e16fd5 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcQueryExecuteRequest.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcQueryExecuteRequest.java
@@ -27,7 +27,7 @@ import org.apache.ignite.internal.tostring.S;
/**
* JDBC query execute request.
*/
-public class JdbcQueryExecuteRequest implements ClientMessage {
+public class JdbcQueryExecuteRequest extends JdbcObservableTimeAwareRequest implements ClientMessage {
/** Expected statement type. */
private JdbcStatementType stmtType;
@@ -61,6 +61,8 @@ public class JdbcQueryExecuteRequest implements ClientMessage {
*/
private long correlationToken;
+ private long observableTime;
+
/**
* Default constructor. For deserialization purposes.
*/
@@ -91,7 +93,8 @@ public class JdbcQueryExecuteRequest implements ClientMessage {
boolean autoCommit,
boolean multiStatement,
long queryTimeoutMillis,
- long correlationToken
+ long correlationToken,
+ long observableTime
) {
Objects.requireNonNull(stmtType);
@@ -105,6 +108,7 @@ public class JdbcQueryExecuteRequest implements ClientMessage {
this.multiStatement = multiStatement;
this.queryTimeoutMillis = queryTimeoutMillis;
this.correlationToken = correlationToken;
+ this.observableTime = observableTime;
}
/**
@@ -190,6 +194,10 @@ public class JdbcQueryExecuteRequest implements ClientMessage {
return correlationToken;
}
+ public long observableTime() {
+ return observableTime;
+ }
+
/** {@inheritDoc} */
@Override
public void writeBinary(ClientMessagePacker packer) {
@@ -204,6 +212,7 @@ public class JdbcQueryExecuteRequest implements ClientMessage {
packer.packObjectArrayAsBinaryTuple(args);
packer.packLong(queryTimeoutMillis);
packer.packLong(correlationToken);
+ packer.packLong(observableTime);
}
/** {@inheritDoc} */
@@ -220,6 +229,7 @@ public class JdbcQueryExecuteRequest implements ClientMessage {
args = unpacker.unpackObjectArrayFromBinaryTuple();
queryTimeoutMillis = unpacker.unpackLong();
correlationToken = unpacker.unpackLong();
+ observableTime = unpacker.unpackLong();
}
/** {@inheritDoc} */
diff --git a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/Response.java b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/Response.java
index 9e625e28279..618e50cb1ca 100644
--- a/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/Response.java
+++ b/modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/Response.java
@@ -39,9 +39,6 @@ public abstract class Response implements ClientMessage {
/** Error. */
private String err;
- /** Has results. */
- protected boolean hasResults;
-
/**
* Constructs successful response.
*/
@@ -121,9 +118,9 @@ public abstract class Response implements ClientMessage {
}
/**
- * Gets hasResults flag.
+ * Gets success status.
*
- * @return Has results.
+ * @return {@code True} if command succeeded, {@code false} otherwise.
*/
public boolean success() {
return status == STATUS_SUCCESS;
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 c7e83e5580b..7aca8ec0db6 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
@@ -714,10 +714,6 @@ public class ClientInboundMessageHandler extends ChannelInboundHandlerAdapter im
return ClientTupleContainsAllKeysRequest.process(in, out, igniteTables, resources, txManager);
case ClientOp.JDBC_CONNECT:
- // TODO: IGNITE-24053 JDBC request ought to contain the client observation timestamp.
- jdbcQueryEventHandler.getTimestampTracker().update(clockService.current());
- out.meta(jdbcQueryEventHandler.getTimestampTracker().get());
-
return ClientJdbcConnectRequest.execute(in, out, jdbcQueryEventHandler);
case ClientOp.JDBC_EXEC:
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcConnectionContext.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcConnectionContext.java
new file mode 100644
index 00000000000..58531845f2e
--- /dev/null
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcConnectionContext.java
@@ -0,0 +1,140 @@
+/*
+ * 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.client.handler;
+
+import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
+
+import java.time.ZoneId;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
+import org.apache.ignite.internal.tx.HybridTimestampTracker;
+import org.apache.ignite.internal.tx.InternalTransaction;
+import org.apache.ignite.internal.tx.InternalTxOptions;
+import org.apache.ignite.internal.tx.TxManager;
+import org.apache.ignite.lang.CancelHandle;
+import org.apache.ignite.lang.CancellationToken;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * JDBC connection context.
+ */
+class JdbcConnectionContext {
+ private final AtomicBoolean closed = new AtomicBoolean();
+
+ private final Object mux = new Object();
+
+ private final TxManager txManager;
+
+ private final ZoneId timeZoneId;
+
+ private final ConcurrentMap<Long, CancelHandle> cancelHandles = new ConcurrentHashMap<>();
+
+ private @Nullable InternalTransaction tx;
+
+ JdbcConnectionContext(
+ TxManager txManager,
+ ZoneId timeZoneId
+ ) {
+ this.txManager = txManager;
+ this.timeZoneId = timeZoneId;
+ }
+
+ ZoneId timeZoneId() {
+ return timeZoneId;
+ }
+
+ /**
+ * Gets the transaction associated with the current connection, starts a new one if it doesn't already exist.
+ *
+ * <p>NOTE: this method is not thread-safe and should only be called by a single thread.
+ *
+ * @return Transaction associated with the current connection.
+ */
+ InternalTransaction getOrStartTransaction() {
+ if (tx == null) {
+ HybridTimestampTracker timeTracker = HybridTimestampTracker.atomicTracker(null);
+
+ tx = txManager.beginExplicitRw(timeTracker, InternalTxOptions.defaults());
+ }
+
+ return tx;
+ }
+
+ /**
+ * Finishes active transaction, if one exists.
+ *
+ * <p>NOTE: this method is not thread-safe and should only be called by a single thread.
+ *
+ * @param commit {@code True} to commit, {@code false} to rollback.
+ * @return Future that represents the pending completion of the operation.
+ */
+ CompletableFuture<@Nullable HybridTimestamp> finishTransactionAsync(boolean commit) {
+ InternalTransaction tx0 = tx;
+
+ tx = null;
+
+ if (tx0 == null) {
+ return nullCompletedFuture();
+ }
+
+ return commit
+ ? tx0.commitAsync().thenApply(ignore -> tx0.observableTimestamp())
+ : tx0.rollbackAsync().thenApply(ignore -> null);
+ }
+
+ boolean valid() {
+ return !closed.get();
+ }
+
+ void close() {
+ if (!closed.compareAndSet(false, true)) {
+ return;
+ }
+
+ synchronized (mux) {
+ finishTransactionAsync(false);
+ }
+ }
+
+ CancellationToken registerExecution(long token) {
+ CancelHandle handle = CancelHandle.create();
+
+ CancelHandle previousHandle = cancelHandles.putIfAbsent(token, handle);
+
+ assert previousHandle == null;
+
+ return handle.token();
+ }
+
+ void deregisterExecution(long token) {
+ cancelHandles.remove(token);
+ }
+
+ CompletableFuture<Void> cancelExecution(long token) {
+ CancelHandle handle = cancelHandles.remove(token);
+
+ if (handle == null) {
+ return nullCompletedFuture();
+ }
+
+ return handle.cancelAsync();
+ }
+}
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcHandlerBase.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcHandlerBase.java
index a3a54a986d3..f1abedb91b1 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcHandlerBase.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcHandlerBase.java
@@ -19,7 +19,6 @@ package org.apache.ignite.client.handler;
import static org.apache.ignite.internal.jdbc.proto.IgniteQueryErrorCode.UNSUPPORTED_OPERATION;
import static org.apache.ignite.internal.sql.engine.SqlQueryType.DDL;
-import static org.apache.ignite.internal.sql.engine.SqlQueryType.DML;
import static org.apache.ignite.internal.sql.engine.SqlQueryType.KILL;
import static org.apache.ignite.internal.sql.engine.SqlQueryType.TX_CONTROL;
@@ -49,16 +48,12 @@ import org.jetbrains.annotations.Nullable;
* Contains common methods used to process jdbc requests.
*/
abstract class JdbcHandlerBase {
-
/** {@link SqlQueryType}s allowed in JDBC select statements. **/
public static final Set<SqlQueryType> SELECT_STATEMENT_QUERIES = Set.of(
SqlQueryType.QUERY,
SqlQueryType.EXPLAIN
);
- /** {@link SqlQueryType}s allowed in JDBC update statements. **/
- public static final Set<SqlQueryType> UPDATE_STATEMENT_QUERIES = Set.of(DML, DDL, KILL);
-
/** {@link SqlQueryType}s types that return 0 in executeUpdate and execute / getUpdateCount. **/
public static final Set<SqlQueryType> ZERO_UPDATE_COUNT_QUERIES = Set.of(DDL, KILL, TX_CONTROL);
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImpl.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImpl.java
index 4614a044236..872a9a62e7a 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImpl.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImpl.java
@@ -21,7 +21,6 @@ import static org.apache.ignite.internal.jdbc.proto.IgniteQueryErrorCode.UNKNOWN
import static org.apache.ignite.internal.sql.engine.SqlQueryType.DML;
import static org.apache.ignite.internal.util.ArrayUtils.OBJECT_EMPTY_ARRAY;
import static org.apache.ignite.internal.util.CompletableFutures.allOf;
-import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.apache.ignite.lang.ErrorGroups.Client.CONNECTION_ERR;
import it.unimi.dsi.fastutil.ints.IntArrayList;
@@ -31,12 +30,10 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Function;
import org.apache.ignite.client.handler.requests.jdbc.JdbcMetadataCatalog;
import org.apache.ignite.client.handler.requests.jdbc.JdbcQueryCursor;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.jdbc.proto.JdbcQueryEventHandler;
import org.apache.ignite.internal.jdbc.proto.JdbcStatementType;
import org.apache.ignite.internal.jdbc.proto.event.JdbcBatchExecuteRequest;
@@ -67,10 +64,8 @@ import org.apache.ignite.internal.sql.engine.property.SqlProperties;
import org.apache.ignite.internal.sql.engine.property.SqlPropertiesHelper;
import org.apache.ignite.internal.tx.HybridTimestampTracker;
import org.apache.ignite.internal.tx.InternalTransaction;
-import org.apache.ignite.internal.tx.InternalTxOptions;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.internal.util.AsyncCursor.BatchedResult;
-import org.apache.ignite.lang.CancelHandle;
import org.apache.ignite.lang.CancellationToken;
import org.jetbrains.annotations.Nullable;
@@ -81,12 +76,6 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
/** {@link SqlQueryType}s allowed in JDBC update statements. **/
public static final Set<SqlQueryType> UPDATE_STATEMENT_QUERIES = Set.of(DML, SqlQueryType.DDL, SqlQueryType.KILL);
- /**
- * Observation timestamp tracker.
- * TODO: IGNITE-24053 Remove this after the issue will be fixed.
- * */
- private final HybridTimestampTracker timestampTracker = HybridTimestampTracker.atomicTracker(null);
-
/** Sql query processor. */
private final QueryProcessor processor;
@@ -158,7 +147,17 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
long correlationToken = req.correlationToken();
CancellationToken token = connectionContext.registerExecution(correlationToken);
- InternalTransaction tx = req.autoCommit() ? null : connectionContext.getOrStartTransaction(timestampTracker);
+ InternalTransaction tx;
+ HybridTimestampTracker timeTracker;
+
+ if (req.autoCommit()) {
+ tx = null;
+ req.observableTimeHolder().set(HybridTimestamp.nullableHybridTimestamp(req.observableTime()));
+ timeTracker = HybridTimestampTracker.fromAtomicReference(req.observableTimeHolder());
+ } else {
+ tx = connectionContext.getOrStartTransaction();
+ timeTracker = HybridTimestampTracker.emptyTracker();
+ }
JdbcStatementType reqStmtType = req.getStmtType();
boolean multiStatement = req.multiStatement();
@@ -169,7 +168,7 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
CompletableFuture<AsyncSqlCursor<InternalSqlRow>> result = processor.queryAsync(
properties,
- timestampTracker,
+ timeTracker,
tx,
token,
req.sqlQuery(),
@@ -182,10 +181,6 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
.exceptionally(t -> createErrorResult("Exception while executing query.", t, null));
}
- public HybridTimestampTracker getTimestampTracker() {
- return timestampTracker;
- }
-
private static SqlProperties createProperties(
JdbcStatementType stmtType,
boolean multiStatement,
@@ -225,7 +220,13 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
return CompletableFuture.completedFuture(new JdbcBatchExecuteResult(Response.STATUS_FAILED, "Connection is broken"));
}
- InternalTransaction tx = req.autoCommit() ? null : connectionContext.getOrStartTransaction(timestampTracker);
+ InternalTransaction tx = req.autoCommit() ? null : connectionContext.getOrStartTransaction();
+ HybridTimestampTracker timeTracker = tx == null
+ ? HybridTimestampTracker.fromAtomicReference(req.observableTimeHolder())
+ : HybridTimestampTracker.emptyTracker();
+
+ assert req.autoCommit() || tx != null;
+
long correlationToken = req.correlationToken();
CancellationToken token = connectionContext.registerExecution(correlationToken);
var queries = req.queries();
@@ -236,6 +237,7 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
for (String query : queries) {
tail = tail.thenCompose(list -> executeAndCollectUpdateCount(
connectionContext,
+ timeTracker,
tx,
token,
query,
@@ -266,7 +268,13 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
return CompletableFuture.completedFuture(new JdbcBatchExecuteResult(Response.STATUS_FAILED, "Connection is broken"));
}
- InternalTransaction tx = req.autoCommit() ? null : connectionContext.getOrStartTransaction(timestampTracker);
+ InternalTransaction tx = req.autoCommit() ? null : connectionContext.getOrStartTransaction();
+ HybridTimestampTracker timeTracker = tx == null
+ ? HybridTimestampTracker.fromAtomicReference(req.observableTimeHolder())
+ : HybridTimestampTracker.emptyTracker();
+
+ assert req.autoCommit() || tx != null;
+
long correlationToken = req.correlationToken();
CancellationToken token = connectionContext.registerExecution(correlationToken);
var argList = req.getArgs();
@@ -276,7 +284,7 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
for (Object[] args : argList) {
tail = tail.thenCompose(list -> executeAndCollectUpdateCount(
- connectionContext, tx, token, req.getQuery(), args, timeoutMillis, list
+ connectionContext, timeTracker, tx, token, req.getQuery(), args, timeoutMillis, list
));
}
@@ -293,6 +301,7 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
private CompletableFuture<IntArrayList> executeAndCollectUpdateCount(
JdbcConnectionContext context,
+ HybridTimestampTracker tracker,
@Nullable InternalTransaction tx,
CancellationToken token,
String sql,
@@ -308,7 +317,7 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
CompletableFuture<AsyncSqlCursor<InternalSqlRow>> result = processor.queryAsync(
properties,
- timestampTracker,
+ tracker,
tx,
token,
sql,
@@ -390,12 +399,12 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
return CompletableFuture.completedFuture(new JdbcFinishTxResult(Response.STATUS_FAILED, "Connection is broken"));
}
- return connectionContext.finishTransactionAsync(commit).handle((ignored, t) -> {
+ return connectionContext.finishTransactionAsync(commit).handle((observableTime, t) -> {
if (t != null) {
return new JdbcFinishTxResult(Response.STATUS_FAILED, t.getMessage());
}
- return new JdbcFinishTxResult();
+ return new JdbcFinishTxResult(observableTime);
});
}
@@ -419,102 +428,6 @@ public class JdbcQueryEventHandlerImpl extends JdbcHandlerBase implements JdbcQu
});
}
- static class JdbcConnectionContext {
- private final AtomicBoolean closed = new AtomicBoolean();
-
- private final Object mux = new Object();
-
- private final TxManager txManager;
-
- private final ZoneId timeZoneId;
-
- private final ConcurrentMap<Long, CancelHandle> cancelHandles = new ConcurrentHashMap<>();
-
- private @Nullable InternalTransaction tx;
-
- JdbcConnectionContext(
- TxManager txManager,
- ZoneId timeZoneId
- ) {
- this.txManager = txManager;
- this.timeZoneId = timeZoneId;
- }
-
- ZoneId timeZoneId() {
- return timeZoneId;
- }
-
- /**
- * Gets the transaction associated with the current connection, starts a new one if it doesn't already exist.
- *
- * <p>NOTE: this method is not thread-safe and should only be called by a single thread.
- *
- * @param timestampProvider Observation timestamp provider.
- * @return Transaction associated with the current connection.
- */
- InternalTransaction getOrStartTransaction(HybridTimestampTracker timestampProvider) {
- return tx == null ? tx = txManager.beginExplicitRw(timestampProvider, InternalTxOptions.defaults()) : tx;
- }
-
- /**
- * Finishes active transaction, if one exists.
- *
- * <p>NOTE: this method is not thread-safe and should only be called by a single thread.
- *
- * @param commit {@code True} to commit, {@code false} to rollback.
- * @return Future that represents the pending completion of the operation.
- */
- CompletableFuture<Void> finishTransactionAsync(boolean commit) {
- InternalTransaction tx0 = tx;
-
- tx = null;
-
- if (tx0 == null) {
- return nullCompletedFuture();
- }
-
- return commit ? tx0.commitAsync() : tx0.rollbackAsync();
- }
-
- boolean valid() {
- return !closed.get();
- }
-
- void close() {
- if (!closed.compareAndSet(false, true)) {
- return;
- }
-
- synchronized (mux) {
- finishTransactionAsync(false);
- }
- }
-
- CancellationToken registerExecution(long token) {
- CancelHandle handle = CancelHandle.create();
-
- CancelHandle previousHandle = cancelHandles.putIfAbsent(token, handle);
-
- assert previousHandle == null;
-
- return handle.token();
- }
-
- void deregisterExecution(long token) {
- cancelHandles.remove(token);
- }
-
- CompletableFuture<Void> cancelExecution(long token) {
- CancelHandle handle = cancelHandles.remove(token);
-
- if (handle == null) {
- return nullCompletedFuture();
- }
-
- return handle.cancelAsync();
- }
- }
-
private void doWhenAllCursorsComplete(
CompletableFuture<AsyncSqlCursor<InternalSqlRow>> cursorFuture, Runnable action
) {
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcExecuteBatchRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcExecuteBatchRequest.java
index d21534938f3..d5ee9d37b5c 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcExecuteBatchRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcExecuteBatchRequest.java
@@ -47,9 +47,7 @@ public class ClientJdbcExecuteBatchRequest {
req.readBinary(in);
return handler.batchAsync(connectionId, req).thenAccept(res -> {
- if (req.autoCommit()) {
- out.meta(handler.getTimestampTracker().get());
- }
+ out.meta(req.observableTimeHolder().get());
res.writeBinary(out);
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcExecuteRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcExecuteRequest.java
index 8e527272df9..1db98e65429 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcExecuteRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcExecuteRequest.java
@@ -41,15 +41,12 @@ public class ClientJdbcExecuteRequest {
JdbcQueryEventHandlerImpl handler
) {
var req = new JdbcQueryExecuteRequest();
-
long connectionId = in.unpackLong();
req.readBinary(in);
return handler.queryAsync(connectionId, req).thenAccept(res -> {
- if (req.autoCommit()) {
- out.meta(handler.getTimestampTracker().get());
- }
+ out.meta(req.observableTimeHolder().get());
res.writeBinary(out);
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcFinishTxRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcFinishTxRequest.java
index b1982a26e6d..7aeb903b3c0 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcFinishTxRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcFinishTxRequest.java
@@ -44,7 +44,7 @@ public class ClientJdbcFinishTxRequest {
return handler.finishTxAsync(connectionId, commit).thenAccept(res -> {
if (commit) {
- out.meta(handler.getTimestampTracker().get());
+ out.meta(res.observableTime());
}
res.writeBinary(out);
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcPreparedStmntBatchRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcPreparedStmntBatchRequest.java
index 4bd46661a35..98bc54b84cc 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcPreparedStmntBatchRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/jdbc/ClientJdbcPreparedStmntBatchRequest.java
@@ -47,9 +47,7 @@ public class ClientJdbcPreparedStmntBatchRequest {
req.readBinary(in);
return handler.batchPrepStatementAsync(connectionId, req).thenAccept(res -> {
- if (req.autoCommit()) {
- out.meta(handler.getTimestampTracker().get());
- }
+ out.meta(req.observableTimeHolder().get());
res.writeBinary(out);
});
diff --git a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteScriptRequest.java b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteScriptRequest.java
index 328d39b1a08..00a67129274 100644
--- a/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteScriptRequest.java
+++ b/modules/client-handler/src/main/java/org/apache/ignite/client/handler/requests/sql/ClientSqlExecuteScriptRequest.java
@@ -50,6 +50,7 @@ public class ClientSqlExecuteScriptRequest {
}
HybridTimestamp clientTs = HybridTimestamp.nullableHybridTimestamp(in.unpackLong());
+ // TODO Thin client script execution is broken - should return observable time to client.
var tsUpdater = HybridTimestampTracker.clientTracker(clientTs, ts -> {});
// TODO https://issues.apache.org/jira/browse/IGNITE-23646 Pass cancellation token to the query processor.
diff --git a/modules/client-handler/src/test/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImplTest.java b/modules/client-handler/src/test/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImplTest.java
index 4df09296b48..76ec0de4530 100644
--- a/modules/client-handler/src/test/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImplTest.java
+++ b/modules/client-handler/src/test/java/org/apache/ignite/client/handler/JdbcQueryEventHandlerImplTest.java
@@ -25,8 +25,10 @@ import static org.apache.ignite.internal.testframework.matchers.CompletableFutur
import static org.apache.ignite.internal.util.CompletableFutures.nullCompletedFuture;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.containsString;
+import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.notNullValue;
+import static org.hamcrest.Matchers.nullValue;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.mock;
@@ -42,28 +44,36 @@ import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Predicate;
-import org.apache.ignite.client.handler.JdbcQueryEventHandlerImpl.JdbcConnectionContext;
import org.apache.ignite.client.handler.requests.jdbc.JdbcMetadataCatalog;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.jdbc.proto.JdbcQueryEventHandler;
import org.apache.ignite.internal.jdbc.proto.JdbcStatementType;
import org.apache.ignite.internal.jdbc.proto.event.JdbcBatchExecuteRequest;
import org.apache.ignite.internal.jdbc.proto.event.JdbcBatchExecuteResult;
import org.apache.ignite.internal.jdbc.proto.event.JdbcBatchPreparedStmntRequest;
import org.apache.ignite.internal.jdbc.proto.event.JdbcConnectResult;
+import org.apache.ignite.internal.jdbc.proto.event.JdbcFinishTxResult;
import org.apache.ignite.internal.jdbc.proto.event.JdbcQueryExecuteRequest;
import org.apache.ignite.internal.jdbc.proto.event.Response;
import org.apache.ignite.internal.lang.IgniteInternalCheckedException;
+import org.apache.ignite.internal.sql.ResultSetMetadataImpl;
import org.apache.ignite.internal.sql.engine.AsyncSqlCursor;
+import org.apache.ignite.internal.sql.engine.AsyncSqlCursorImpl;
import org.apache.ignite.internal.sql.engine.InternalSqlRow;
import org.apache.ignite.internal.sql.engine.QueryCancelledException;
import org.apache.ignite.internal.sql.engine.QueryProcessor;
+import org.apache.ignite.internal.sql.engine.SqlQueryType;
+import org.apache.ignite.internal.sql.engine.util.IteratorToDataCursorAdapter;
import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.apache.ignite.internal.tx.HybridTimestampTracker;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.TxManager;
import org.apache.ignite.lang.CancelHandleHelper;
import org.apache.ignite.lang.CancellationToken;
+import org.apache.ignite.sql.ResultSetMetadata;
import org.hamcrest.CustomMatcher;
import org.hamcrest.Matcher;
import org.junit.jupiter.api.BeforeEach;
@@ -136,11 +146,11 @@ class JdbcQueryEventHandlerImplTest extends BaseIgniteAbstractTest {
registryCloseLatch.countDown();
assertThat(startTxLatch.await(timeout, TimeUnit.SECONDS), is(true));
- return null;
+ return mock(InternalTransaction.class);
});
CompletableFuture<JdbcBatchExecuteResult> fut = IgniteTestUtils.runAsync(() ->
- eventHandler.batchAsync(connectionId, createExecuteBatchRequest("x", "QUERY")).get()
+ eventHandler.batchAsync(connectionId, createExecuteBatchRequest("x", false, "QUERY")).get()
);
assertThat(registryCloseLatch.await(timeout, TimeUnit.SECONDS), is(true));
@@ -164,7 +174,7 @@ class JdbcQueryEventHandlerImplTest extends BaseIgniteAbstractTest {
long connectionId = acquireConnectionId();
- await(eventHandler.batchAsync(connectionId, createExecuteBatchRequest("x", "UPDATE 1")));
+ await(eventHandler.batchAsync(connectionId, createExecuteBatchRequest("x", false, "UPDATE 1")));
verify(txManager).beginExplicitRw(any(), any());
verify(tx, times(0)).rollbackAsync();
@@ -191,22 +201,22 @@ class JdbcQueryEventHandlerImplTest extends BaseIgniteAbstractTest {
JdbcStatementType type = JdbcStatementType.SELECT_STATEMENT_TYPE;
await(eventHandler.queryAsync(
- connectionId, createExecuteRequest(schema, "SELECT 1", type)
+ connectionId, createExecuteRequest(schema, "SELECT 1", type, false)
));
verify(txManager, times(1)).beginExplicitRw(any(), any());
- await(eventHandler.batchAsync(connectionId, createExecuteBatchRequest("schema", "UPDATE 1", "UPDATE 2")));
+ await(eventHandler.batchAsync(connectionId, createExecuteBatchRequest("schema", false, "UPDATE 1", "UPDATE 2")));
verify(txManager, times(1)).beginExplicitRw(any(), any());
await(eventHandler.finishTxAsync(connectionId, false));
verify(tx).rollbackAsync();
- await(eventHandler.batchAsync(connectionId, createExecuteBatchRequest("schema", "UPDATE 1", "UPDATE 2")));
+ await(eventHandler.batchAsync(connectionId, createExecuteBatchRequest("schema", false, "UPDATE 1", "UPDATE 2")));
verify(txManager, times(2)).beginExplicitRw(any(), any());
await(eventHandler.queryAsync(
- connectionId, createExecuteRequest(schema, "SELECT 2", type)
+ connectionId, createExecuteRequest(schema, "SELECT 2", type, false)
));
verify(txManager, times(2)).beginExplicitRw(any(), any());
- await(eventHandler.batchAsync(connectionId, createExecuteBatchRequest("schema", "UPDATE 3", "UPDATE 4")));
+ await(eventHandler.batchAsync(connectionId, createExecuteBatchRequest("schema", false, "UPDATE 3", "UPDATE 4")));
verify(txManager, times(2)).beginExplicitRw(any(), any());
await(eventHandler.finishTxAsync(connectionId, true));
@@ -222,7 +232,7 @@ class JdbcQueryEventHandlerImplTest extends BaseIgniteAbstractTest {
long connectionId = acquireConnectionId();
- JdbcQueryExecuteRequest executeRequest = createExecuteRequest("schema", "SELECT 1", JdbcStatementType.SELECT_STATEMENT_TYPE);
+ JdbcQueryExecuteRequest executeRequest = createExecuteRequest("schema", "SELECT 1", JdbcStatementType.SELECT_STATEMENT_TYPE, true);
CompletableFuture<? extends Response> resultFuture = eventHandler.queryAsync(connectionId, executeRequest);
@@ -237,7 +247,7 @@ class JdbcQueryEventHandlerImplTest extends BaseIgniteAbstractTest {
long connectionId = acquireConnectionId();
- JdbcBatchExecuteRequest executeRequest = createExecuteBatchRequest("schema", "SELECT 1");
+ JdbcBatchExecuteRequest executeRequest = createExecuteBatchRequest("schema", true, "SELECT 1");
CompletableFuture<? extends Response> resultFuture = eventHandler.batchAsync(connectionId, executeRequest);
@@ -261,6 +271,147 @@ class JdbcQueryEventHandlerImplTest extends BaseIgniteAbstractTest {
cancelAndVerify(connectionId, executeRequest.correlationToken(), resultFuture);
}
+ @Test
+ void handlerTracksObservableTimeFromClient() {
+ HybridTimestamp clientTs = HybridTimestamp.hybridTimestamp(200L);
+ HybridTimestamp updatedTs = HybridTimestamp.hybridTimestamp(300L);
+ HybridTimestamp ignoredTs = HybridTimestamp.hybridTimestamp(100L);
+ HybridTimestamp txTime = HybridTimestamp.hybridTimestamp(50L);
+
+ long connectionId = acquireConnectionId();
+
+ AtomicReference<HybridTimestamp> fromClient = new AtomicReference<>();
+
+ HybridTimestampTracker txTracker = HybridTimestampTracker.atomicTracker(txTime);
+ InternalTransaction rwTx = mock(InternalTransaction.class);
+ when(rwTx.commitAsync()).thenReturn(nullCompletedFuture());
+ when(rwTx.rollbackAsync()).thenReturn(nullCompletedFuture());
+ when(rwTx.observableTimestamp()).thenReturn(txTracker.get());
+ when(txManager.beginExplicitRw(any(), any())).thenReturn(rwTx);
+
+ when(queryProcessor.queryAsync(any(), any(HybridTimestampTracker.class), any(), any(), any(), any(Object[].class)))
+ .thenAnswer(invocation -> {
+ HybridTimestampTracker tracker = invocation.getArgument(1);
+ InternalTransaction explicitTx = invocation.getArgument(2);
+
+ if (explicitTx == null || explicitTx.isReadOnly()) {
+ fromClient.set(tracker.get());
+
+ tracker.update(updatedTs);
+
+ // Time should only increase, smaller values should be ignored.
+ tracker.update(ignoredTs);
+ }
+
+ return CompletableFuture.completedFuture(createEmptyQueryCursor());
+ });
+
+ // Query (autocommit).
+ {
+ JdbcQueryExecuteRequest executeRequest = new JdbcQueryExecuteRequest(
+ JdbcStatementType.SELECT_STATEMENT_TYPE,
+ "schema",
+ 1024,
+ 1,
+ "SELECT 1",
+ new Object[0],
+ true,
+ false,
+ 0,
+ System.currentTimeMillis(),
+ clientTs.longValue()
+ );
+
+ await(eventHandler.queryAsync(connectionId, executeRequest));
+
+ assertThat("Must take into account the value from the request",
+ fromClient.get(), equalTo(clientTs));
+ assertThat("Should return the updated tracker value to the client.",
+ executeRequest.observableTimeHolder().get(), equalTo(updatedTs));
+ }
+
+ fromClient.set(null);
+
+ // Batch (autocommit).
+ {
+ JdbcBatchExecuteRequest executeRequest = createExecuteBatchRequest("schema", true, "SELECT 1");
+
+ await(eventHandler.batchAsync(connectionId, executeRequest));
+
+ assertThat("No value should be passed from the client.",
+ fromClient.get(), nullValue());
+ assertThat("Should return the updated tracker value to the client.",
+ executeRequest.observableTimeHolder().get(), equalTo(updatedTs));
+ }
+
+ // Prepared batch (autocommit).
+ {
+ JdbcBatchPreparedStmntRequest executeRequest = createExecutePrepBatchRequest("schema", "SELECT 1");
+
+ await(eventHandler.batchPrepStatementAsync(connectionId, executeRequest));
+
+ assertThat("No value should be passed from the client.",
+ fromClient.get(), nullValue());
+ assertThat("Should return the updated tracker value to the client.",
+ executeRequest.observableTimeHolder().get(), equalTo(updatedTs));
+ }
+
+ // Query (non-autocommit).
+ {
+ JdbcQueryExecuteRequest executeRequest = new JdbcQueryExecuteRequest(
+ JdbcStatementType.SELECT_STATEMENT_TYPE,
+ "schema",
+ 1024,
+ 1,
+ "SELECT 1",
+ new Object[0],
+ false,
+ false,
+ 0,
+ System.currentTimeMillis(),
+ clientTs.longValue()
+ );
+
+ await(eventHandler.queryAsync(connectionId, executeRequest));
+
+ assertThat("Should not update observable time until tx commit.",
+ executeRequest.observableTimeHolder().get(), nullValue());
+
+ // Rollback.
+ JdbcFinishTxResult res = await(eventHandler.finishTxAsync(connectionId, false));
+ assertThat("The observable time must not be returned when rolling back a tx.", res.observableTime(), nullValue());
+ verify(rwTx).rollbackAsync();
+ }
+
+ // Batch (non-autocommit).
+ {
+ JdbcBatchExecuteRequest executeRequest = createExecuteBatchRequest("schema", false, "SELECT 1");
+
+ await(eventHandler.batchAsync(connectionId, executeRequest));
+
+ assertThat("Should not update observable time until tx commit.",
+ executeRequest.observableTimeHolder().get(), nullValue());
+ }
+
+ // Prepared batch (non-autocommit).
+ {
+ JdbcBatchPreparedStmntRequest executeRequest = new JdbcBatchPreparedStmntRequest(
+ "schema", "SELECT 1", Collections.singletonList(new Object[] {1}), false, 0, System.currentTimeMillis());
+
+ await(eventHandler.batchPrepStatementAsync(connectionId, executeRequest));
+
+ assertThat("Should not update observable time until tx commit.",
+ executeRequest.observableTimeHolder().get(), nullValue());
+ }
+
+ // Commit.
+ {
+ JdbcFinishTxResult res = await(eventHandler.finishTxAsync(connectionId, true));
+ assertThat(res.observableTime(), equalTo(txTime));
+ verify(rwTx).commitAsync();
+ }
+ }
+
private long acquireConnectionId() {
JdbcConnectResult result = await(eventHandler.connect(ZoneId.systemDefault()));
@@ -298,22 +449,22 @@ class JdbcQueryEventHandlerImplTest extends BaseIgniteAbstractTest {
);
}
- private static JdbcQueryExecuteRequest createExecuteRequest(String schema, String query, JdbcStatementType type) {
+ private static JdbcQueryExecuteRequest createExecuteRequest(String schema, String query, JdbcStatementType type, boolean autoCommit) {
//noinspection DataFlowIssue
return new JdbcQueryExecuteRequest(
- type, schema, 1024, 1, query, null, false, false, 0, System.currentTimeMillis()
+ type, schema, 1024, 1, query, null, autoCommit, false, 0, System.currentTimeMillis(), 0L
);
}
- private static JdbcBatchExecuteRequest createExecuteBatchRequest(String schema, String... statements) {
+ private static JdbcBatchExecuteRequest createExecuteBatchRequest(String schema, boolean autoCommit, String... statements) {
return new JdbcBatchExecuteRequest(
- schema, List.of(statements), false, 0, System.currentTimeMillis()
+ schema, List.of(statements), autoCommit, 0, System.currentTimeMillis()
);
}
private static JdbcBatchPreparedStmntRequest createExecutePrepBatchRequest(String schema, String query) {
return new JdbcBatchPreparedStmntRequest(
- schema, query, Collections.singletonList(new Object[] {1}), false, 0, System.currentTimeMillis()
+ schema, query, Collections.singletonList(new Object[] {1}), true, 0, System.currentTimeMillis()
);
}
@@ -325,4 +476,15 @@ class JdbcQueryEventHandlerImplTest extends BaseIgniteAbstractTest {
}
};
}
+
+ private static AsyncSqlCursor<InternalSqlRow> createEmptyQueryCursor() {
+ ResultSetMetadata emptyMeta = new ResultSetMetadataImpl(List.of());
+
+ return new AsyncSqlCursorImpl<>(
+ SqlQueryType.QUERY,
+ emptyMeta,
+ new IteratorToDataCursorAdapter<>(Collections.emptyIterator()),
+ null
+ );
+ }
}
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 1dc83e40f03..e5b831b3fab 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
@@ -115,7 +115,7 @@ public final class ReliableChannel implements AutoCloseable {
private final AtomicLong partitionAssignmentTimestamp = new AtomicLong();
/** Observable timestamp, or causality token. Sent by the server with every response, and required by some requests. */
- private final AtomicLong observableTimestamp = new AtomicLong();
+ private final AtomicLong observableTimestamp;
/** Cluster id from the first handshake. */
private final AtomicReference<UUID> clusterId = new AtomicReference<>();
@@ -129,15 +129,19 @@ public final class ReliableChannel implements AutoCloseable {
*
* @param chFactory Channel factory.
* @param clientCfg Client config.
+ * @param metrics Client metrics.
+ * @param observableTime Tracker of the latest time observed by client.
*/
ReliableChannel(
ClientChannelFactory chFactory,
IgniteClientConfiguration clientCfg,
- ClientMetricSource metrics) {
+ ClientMetricSource metrics,
+ AtomicLong observableTime) {
this.clientCfg = Objects.requireNonNull(clientCfg, "clientCfg");
this.chFactory = Objects.requireNonNull(chFactory, "chFactory");
this.log = ClientUtils.logger(clientCfg, ReliableChannel.class);
this.metrics = metrics;
+ this.observableTimestamp = Objects.requireNonNull(observableTime, "observableTime");
connMgr = new NettyClientConnectionMultiplexer(metrics);
connMgr.start(clientCfg);
diff --git a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
index 0de3774a83b..4e9a44f7c62 100644
--- a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
+++ b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
@@ -25,6 +25,7 @@ import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicLong;
import org.apache.ignite.catalog.IgniteCatalog;
import org.apache.ignite.client.IgniteClient;
import org.apache.ignite.client.IgniteClientConfiguration;
@@ -89,13 +90,15 @@ public class TcpIgniteClient implements IgniteClient {
*/
private String clusterName;
+
/**
* Constructor.
*
* @param cfg Config.
+ * @param observableTime Tracker of the latest time observed by client.
*/
- private TcpIgniteClient(IgniteClientConfiguration cfg) {
- this(TcpClientChannel::createAsync, cfg);
+ private TcpIgniteClient(IgniteClientConfiguration cfg, AtomicLong observableTime) {
+ this(TcpClientChannel::createAsync, cfg, observableTime);
}
/**
@@ -103,15 +106,16 @@ public class TcpIgniteClient implements IgniteClient {
*
* @param chFactory Channel factory.
* @param cfg Config.
+ * @param observableTime Tracker of the latest time observed by client.
*/
- private TcpIgniteClient(ClientChannelFactory chFactory, IgniteClientConfiguration cfg) {
+ private TcpIgniteClient(ClientChannelFactory chFactory, IgniteClientConfiguration cfg, @Nullable AtomicLong observableTime) {
assert chFactory != null;
assert cfg != null;
this.cfg = cfg;
metrics = new ClientMetricSource();
- ch = new ReliableChannel(chFactory, cfg, metrics);
+ ch = new ReliableChannel(chFactory, cfg, metrics, observableTime);
tables = new ClientTables(ch, marshallers);
transactions = new ClientTransactions(ch);
compute = new ClientCompute(ch, tables);
@@ -154,11 +158,21 @@ public class TcpIgniteClient implements IgniteClient {
* @return Future representing pending completion of the operation.
*/
public static CompletableFuture<IgniteClient> startAsync(IgniteClientConfiguration cfg) {
+ return startAsync(cfg, new AtomicLong());
+ }
+
+ /**
+ * Initializes new instance of {@link IgniteClient} and establishes the connection.
+ *
+ * @param cfg Thin client configuration.
+ * @return Future representing pending completion of the operation.
+ */
+ public static CompletableFuture<IgniteClient> startAsync(IgniteClientConfiguration cfg, AtomicLong tracker) {
ErrorGroups.initialize();
try {
//noinspection resource: returned from method
- var client = new TcpIgniteClient(cfg);
+ var client = new TcpIgniteClient(cfg, tracker);
return client.initAsync().thenApply(x -> client);
} catch (IgniteException e) {
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 8cefbecb722..0f5cee2324a 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
@@ -150,6 +150,11 @@ public class FakeTxManager implements TxManager {
return timestamp;
}
+ @Override
+ public @Nullable HybridTimestamp observableTimestamp() {
+ return tracker.get();
+ }
+
@Override
public boolean implicit() {
return false;
diff --git a/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcMultipleConnectionsTest.java b/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcMultipleConnectionsTest.java
new file mode 100644
index 00000000000..f2d329fc03a
--- /dev/null
+++ b/modules/jdbc/src/integrationTest/java/org/apache/ignite/jdbc/ItJdbcMultipleConnectionsTest.java
@@ -0,0 +1,84 @@
+/*
+ * 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.jdbc;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Multiple JDBC connections test.
+ */
+public class ItJdbcMultipleConnectionsTest extends AbstractJdbcSelfTest {
+ private static final String URL1 = "jdbc:ignite:thin://127.0.0.1:10800";
+ private static final String URL2 = "jdbc:ignite:thin://127.0.0.1:10801";
+
+ @Override
+ protected int initialNodes() {
+ return 2;
+ }
+
+ @AfterEach
+ public void dropTable() {
+ sql("DROP TABLE IF EXISTS T1");
+ }
+
+ @Test
+ public void testMultipleConnectionsSingleServer() throws SQLException {
+ try (Connection conn = DriverManager.getConnection(URL1)) {
+ checkUpdatesVisibility(conn, conn);
+ }
+ }
+
+ @Test
+ public void testMultipleConnectionsMultipleServer() throws SQLException {
+ try (Connection conn1 = DriverManager.getConnection(URL1)) {
+ try (Connection conn2 = DriverManager.getConnection(URL2)) {
+ checkUpdatesVisibility(conn1, conn2);
+ }
+ }
+ }
+
+ private static void checkUpdatesVisibility(Connection conn1, Connection conn2) throws SQLException {
+ try (Statement stmt1 = conn1.createStatement()) {
+ try (Statement stmt2 = conn2.createStatement()) {
+ stmt1.executeUpdate("CREATE TABLE T1(id INT PRIMARY KEY)");
+
+ int rowsCount = 100;
+ int i = 0;
+
+ do {
+ assertThat(stmt1.executeUpdate("INSERT INTO T1 VALUES (" + i + ")"), is(1));
+
+ try (ResultSet rs = stmt2.executeQuery("SELECT count(*) FROM T1 WHERE id >= 0")) {
+ assertTrue(rs.next());
+ assertThat("i=" + i, rs.getLong(1), is(i + 1L));
+ }
+ } while (i++ < rowsCount);
+ }
+ }
+ }
+}
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 779942a7189..15aa9576bb4 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
@@ -23,6 +23,7 @@ import static java.sql.ResultSet.HOLD_CURSORS_OVER_COMMIT;
import static java.sql.ResultSet.TYPE_FORWARD_ONLY;
import static org.apache.ignite.internal.jdbc.proto.SqlStateCode.CLIENT_CONNECTION_FAILED;
import static org.apache.ignite.internal.jdbc.proto.SqlStateCode.CONNECTION_CLOSED;
+import static org.apache.ignite.internal.util.ViewUtils.sync;
import java.sql.Array;
import java.sql.Blob;
@@ -54,10 +55,11 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.ignite.client.BasicAuthenticator;
-import org.apache.ignite.client.IgniteClient;
import org.apache.ignite.client.IgniteClientAuthenticator;
+import org.apache.ignite.client.IgniteClientConfiguration;
import org.apache.ignite.client.SslConfiguration;
import org.apache.ignite.internal.client.HostAndPort;
+import org.apache.ignite.internal.client.IgniteClientConfigurationImpl;
import org.apache.ignite.internal.client.TcpIgniteClient;
import org.apache.ignite.internal.jdbc.proto.IgniteQueryErrorCode;
import org.apache.ignite.internal.jdbc.proto.JdbcQueryEventHandler;
@@ -127,7 +129,7 @@ public class JdbcConnection implements Connection {
*
* @param props Connection properties.
*/
- public JdbcConnection(ConnectionProperties props) throws SQLException {
+ public JdbcConnection(ConnectionProperties props, AtomicLong observableTimeTracker) throws SQLException {
this.connProps = props;
autoCommit = true;
@@ -138,13 +140,7 @@ public class JdbcConnection implements Connection {
qryTimeout = connProps.getQueryTimeout();
try {
- client = ((TcpIgniteClient) IgniteClient.builder()
- .addresses(addrs)
- .connectTimeout(netTimeout)
- .ssl(extractSslConfiguration(connProps))
- .authenticator(extractAuthenticationConfiguration(connProps))
- .build());
-
+ client = buildClient(addrs, observableTimeTracker);
} catch (Exception e) {
throw new SQLException("Failed to connect to server", CLIENT_CONNECTION_FAILED, e);
}
@@ -174,6 +170,25 @@ public class JdbcConnection implements Connection {
holdability = HOLD_CURSORS_OVER_COMMIT;
}
+ private TcpIgniteClient buildClient(String[] addrs, AtomicLong observableTime) {
+ var cfg = new IgniteClientConfigurationImpl(
+ null,
+ addrs,
+ netTimeout,
+ IgniteClientConfigurationImpl.DFLT_BACKGROUND_RECONNECT_INTERVAL,
+ null,
+ IgniteClientConfigurationImpl.DFLT_HEARTBEAT_INTERVAL,
+ IgniteClientConfigurationImpl.DFLT_HEARTBEAT_TIMEOUT,
+ null,
+ null,
+ extractSslConfiguration(connProps),
+ false,
+ extractAuthenticationConfiguration(connProps),
+ IgniteClientConfiguration.DFLT_OPERATION_TIMEOUT);
+
+ return (TcpIgniteClient) sync(TcpIgniteClient.startAsync(cfg, observableTime));
+ }
+
/**
* Constructor used for testing purposes.
*/
@@ -848,6 +863,11 @@ public class JdbcConnection implements Connection {
return connectionId;
}
+ /** Returns an observable timestamp. */
+ long observableTimestamp() {
+ return client.channel().observableTimestamp();
+ }
+
/** {@inheritDoc} */
@Override
public <T> T unwrap(Class<T> iface) throws SQLException {
diff --git a/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcStatement.java b/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcStatement.java
index 48270bc32cb..e5b2507a58a 100644
--- a/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcStatement.java
+++ b/modules/jdbc/src/main/java/org/apache/ignite/internal/jdbc/JdbcStatement.java
@@ -143,9 +143,10 @@ public class JdbcStatement implements Statement {
long correlationToken = nextToken();
JdbcQueryExecuteRequest req = new JdbcQueryExecuteRequest(stmtType, schema, pageSize, maxRows, sql, args,
- conn.getAutoCommit(), multiStatement, queryTimeoutMillis, correlationToken);
+ conn.getAutoCommit(), multiStatement, queryTimeoutMillis, correlationToken, conn.observableTimestamp());
JdbcQueryExecuteResponse res;
+
try {
res = (JdbcQueryExecuteResponse) conn.handler().queryAsync(conn.connectionId(), req).get();
} catch (InterruptedException e) {
diff --git a/modules/jdbc/src/main/java/org/apache/ignite/jdbc/IgniteJdbcDriver.java b/modules/jdbc/src/main/java/org/apache/ignite/jdbc/IgniteJdbcDriver.java
index fd935978f39..45bc1ed33ca 100644
--- a/modules/jdbc/src/main/java/org/apache/ignite/jdbc/IgniteJdbcDriver.java
+++ b/modules/jdbc/src/main/java/org/apache/ignite/jdbc/IgniteJdbcDriver.java
@@ -27,6 +27,7 @@ import java.sql.DriverPropertyInfo;
import java.sql.SQLException;
import java.sql.SQLFeatureNotSupportedException;
import java.util.Properties;
+import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Logger;
import org.apache.ignite.internal.client.proto.ProtocolVersion;
import org.apache.ignite.internal.jdbc.ConnectionPropertiesImpl;
@@ -161,6 +162,8 @@ public class IgniteJdbcDriver implements Driver {
/** Minor version. */
private static final int MINOR_VER = ProtocolVersion.LATEST_VER.minor();
+ private final AtomicLong observableTimeTracker = new AtomicLong();
+
/** {@inheritDoc} */
@Override
public Connection connect(String url, Properties props) throws SQLException {
@@ -172,7 +175,7 @@ public class IgniteJdbcDriver implements Driver {
connProps.init(url, props);
- return new JdbcConnection(connProps);
+ return new JdbcConnection(connProps, observableTimeTracker);
}
/** {@inheritDoc} */
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 afca475a9bb..4d44c7ea6b2 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
@@ -33,6 +33,7 @@ import org.apache.ignite.internal.tx.TxState;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.network.NetworkAddress;
import org.apache.ignite.tx.TransactionException;
+import org.jetbrains.annotations.Nullable;
/**
* Dummy transaction that should be used as mock transaction for execution tests.
@@ -132,6 +133,11 @@ public final class NoOpTransaction implements InternalTransaction {
return hybridTimestamp;
}
+ @Override
+ public @Nullable HybridTimestamp observableTimestamp() {
+ return null;
+ }
+
@Override
public UUID id() {
return id;
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/HybridTimestampTracker.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/HybridTimestampTracker.java
index ec994d78ab8..2d53cc32c08 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/HybridTimestampTracker.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/HybridTimestampTracker.java
@@ -21,6 +21,7 @@ import static org.apache.ignite.internal.hlc.HybridTimestamp.hybridTimestampToLo
import static org.apache.ignite.internal.hlc.HybridTimestamp.nullableHybridTimestamp;
import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.jetbrains.annotations.Nullable;
@@ -96,6 +97,29 @@ public interface HybridTimestampTracker {
};
}
+ /**
+ * Creates an atomic HybridTimestampTracker instance that uses external {@link AtomicReference} to track and update the timestamp.
+ *
+ * @param timestampHolder Timestamp holder.
+ * @return A HybridTimestampTracker instance that uses the provided initial timestamp and update mechanism.
+ */
+ static HybridTimestampTracker fromAtomicReference(AtomicReference<HybridTimestamp> timestampHolder) {
+ return new HybridTimestampTracker() {
+ /** Timestamp. */
+ private final AtomicReference<HybridTimestamp> timestamp = timestampHolder;
+
+ @Override
+ public @Nullable HybridTimestamp get() {
+ return timestamp.get();
+ }
+
+ @Override
+ public void update(HybridTimestamp ts) {
+ timestamp.updateAndGet(x -> (x == null || x.compareTo(ts) < 0) ? ts : x);
+ }
+ };
+ }
+
/**
* Get the observable timestamp.
*
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 205717c8d23..75a2b42ad42 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
@@ -91,6 +91,13 @@ public interface InternalTransaction extends Transaction {
*/
HybridTimestamp startTimestamp();
+ /**
+ * Returns observable timestamp.
+ *
+ * @return observable timestamp.
+ */
+ @Nullable HybridTimestamp observableTimestamp();
+
/**
* Get the transaction coordinator inconsistent ID.
*
diff --git a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/IgniteAbstractTransactionImpl.java b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/IgniteAbstractTransactionImpl.java
index b2dcbe58c89..0cf0068d497 100644
--- a/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/IgniteAbstractTransactionImpl.java
+++ b/modules/transactions/src/main/java/org/apache/ignite/internal/tx/impl/IgniteAbstractTransactionImpl.java
@@ -26,6 +26,7 @@ import static org.apache.ignite.lang.ErrorGroups.Transactions.TX_ROLLBACK_ERR;
import java.util.UUID;
import java.util.concurrent.ExecutionException;
+import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.tx.HybridTimestampTracker;
import org.apache.ignite.internal.tx.InternalTransaction;
import org.apache.ignite.internal.tx.TxManager;
@@ -135,6 +136,12 @@ public abstract class IgniteAbstractTransactionImpl implements InternalTransacti
}
}
+ /** {@inheritDoc} */
+ @Override
+ public @Nullable HybridTimestamp observableTimestamp() {
+ return observableTsTracker.get();
+ }
+
// TODO: remove after IGNITE-22721 gets resolved.
private static Throwable tryToCopyExceptionWithCause(ExecutionException exception) {
Throwable copy = copyExceptionWithCause(exception);
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 6560086411e..35ecce9a8a5 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
@@ -121,6 +121,11 @@ public class PublicApiThreadingTransaction implements InternalTransaction, Wrapp
return transaction.startTimestamp();
}
+ @Override
+ public HybridTimestamp observableTimestamp() {
+ return transaction.observableTimestamp();
+ }
+
@Override
public UUID coordinatorId() {
return transaction.coordinatorId();