IGNITE-24053 Code cleanup (wip).

modules/client-common/src/main/java/org/apache/ignite/internal/jdbc/proto/event/JdbcQueryExecuteRequest.java

@@ -20,6 +20,7 @@
 import java.util.Objects;
 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.jdbc.proto.ClientMessage;
 import org.apache.ignite.internal.jdbc.proto.JdbcStatementType;
 import org.apache.ignite.internal.tostring.S;
@@ -230,6 +231,8 @@ public void readBinary(ClientMessageUnpacker unpacker) {
         queryTimeoutMillis = unpacker.unpackLong();
         correlationToken = unpacker.unpackLong();
         observableTime = unpacker.unpackLong();
+
+        observableTimeUpdatesHolder().set(HybridTimestamp.nullableHybridTimestamp(observableTime));
     }
 
     /** {@inheritDoc} */

modules/client-handler/src/main/java/org/apache/ignite/client/handler/JdbcConnectionContext.java

@@ -0,0 +1,133 @@
+/*
+ * 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.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() {
+        return tx == null ? tx = txManager.begin(HybridTimestampTracker.atomicTracker(null), false) : 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.observableTsTracker().get())
+                : 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();
+    }
+}