From 9644ab2921a8f981c82968973cf33323325c9792 Mon Sep 17 00:00:00 2001
From: Laird Nelson <laird.nelson@oracle.com>
Date: Fri, 20 Jan 2023 10:07:59 -0800
Subject: [PATCH] Introduces JtaConnection.java (#5905)

Signed-off-by: Laird Nelson <laird.nelson@oracle.com>
---
 .../jdbc/DelegatingPreparedStatement.java     |    3 +-
 .../integrations/jdbc/SQLSupplier.java        |   39 +
 .../jta/jdbc/etc/spotbugs/exclude.xml         |    2 +-
 .../jta/jdbc/JtaAdaptingDataSource.java       |  250 ++++
 .../integrations/jta/jdbc/JtaConnection.java  | 1214 +++++++++++++++++
 .../jta/jdbc/TestJtaConnection.java           |  407 ++++++
 .../TestTransactionSpecificConnection.java    |  165 +++
 7 files changed, 2078 insertions(+), 2 deletions(-)
 create mode 100644 integrations/jdbc/jdbc/src/main/java/io/helidon/integrations/jdbc/SQLSupplier.java
 create mode 100644 integrations/jta/jdbc/src/main/java/io/helidon/integrations/jta/jdbc/JtaAdaptingDataSource.java
 create mode 100644 integrations/jta/jdbc/src/main/java/io/helidon/integrations/jta/jdbc/JtaConnection.java
 create mode 100644 integrations/jta/jdbc/src/test/java/io/helidon/integrations/jta/jdbc/TestJtaConnection.java
 create mode 100644 integrations/jta/jdbc/src/test/java/io/helidon/integrations/jta/jdbc/TestTransactionSpecificConnection.java

diff --git a/integrations/jdbc/jdbc/src/main/java/io/helidon/integrations/jdbc/DelegatingPreparedStatement.java b/integrations/jdbc/jdbc/src/main/java/io/helidon/integrations/jdbc/DelegatingPreparedStatement.java
index 611a6ae95f5..4a34512fc49 100644
--- a/integrations/jdbc/jdbc/src/main/java/io/helidon/integrations/jdbc/DelegatingPreparedStatement.java
+++ b/integrations/jdbc/jdbc/src/main/java/io/helidon/integrations/jdbc/DelegatingPreparedStatement.java
@@ -1,5 +1,5 @@
 /*
- * Copyright (c) 2022 Oracle and/or its affiliates.
+ * Copyright (c) 2022, 2023 Oracle and/or its affiliates.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -61,6 +61,7 @@ public class DelegatingPreparedStatement<S extends PreparedStatement> extends De
      * @param strictClosedChecking if {@code true}, then <em>this</em> {@link DelegatingPreparedStatement}'s {@link
      * #isClosed()} method will be invoked before every operation that cannot take place on a closed statement, and, if
      * it returns {@code true}, the operation in question will fail with a {@link SQLException}
+     *
      * @exception NullPointerException if either argument is {@code
      * null}
      *
diff --git a/integrations/jdbc/jdbc/src/main/java/io/helidon/integrations/jdbc/SQLSupplier.java b/integrations/jdbc/jdbc/src/main/java/io/helidon/integrations/jdbc/SQLSupplier.java
new file mode 100644
index 00000000000..39b1f06db4d
--- /dev/null
+++ b/integrations/jdbc/jdbc/src/main/java/io/helidon/integrations/jdbc/SQLSupplier.java
@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) 2023 Oracle and/or its affiliates.
+ *
+ * Licensed 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 io.helidon.integrations.jdbc;
+
+import java.sql.SQLException;
+
+/**
+ * A useful functional interface whose implementations can perform work that may throw a {@link SQLException}.
+ *
+ * @param <T> the type of the object supplied
+ *
+ * @see #get()
+ */
+@FunctionalInterface
+public interface SQLSupplier<T> {
+
+    /**
+     * Performs work and returns the result.
+     *
+     * @return the result of the work
+     *
+     * @exception SQLException if a database access error occurs
+     */
+    T get() throws SQLException;
+
+}
diff --git a/integrations/jta/jdbc/etc/spotbugs/exclude.xml b/integrations/jta/jdbc/etc/spotbugs/exclude.xml
index 4d51b840347..970ca624b0f 100644
--- a/integrations/jta/jdbc/etc/spotbugs/exclude.xml
+++ b/integrations/jta/jdbc/etc/spotbugs/exclude.xml
@@ -1,6 +1,6 @@
 <?xml version="1.0" encoding="UTF-8"?>
 <!--
-    Copyright (c) 2022 Oracle and/or its affiliates.
+    Copyright (c) 2022, 2023 Oracle and/or its affiliates.
 
     Licensed under the Apache License, Version 2.0 (the "License");
     you may not use this file except in compliance with the License.
diff --git a/integrations/jta/jdbc/src/main/java/io/helidon/integrations/jta/jdbc/JtaAdaptingDataSource.java b/integrations/jta/jdbc/src/main/java/io/helidon/integrations/jta/jdbc/JtaAdaptingDataSource.java
new file mode 100644
index 00000000000..5fc7c6d70af
--- /dev/null
+++ b/integrations/jta/jdbc/src/main/java/io/helidon/integrations/jta/jdbc/JtaAdaptingDataSource.java
@@ -0,0 +1,250 @@
+/*
+ * Copyright (c) 2023 Oracle and/or its affiliates.
+ *
+ * Licensed 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 io.helidon.integrations.jta.jdbc;
+
+import java.sql.Connection;
+import java.sql.SQLException;
+import java.util.Objects;
+
+import javax.sql.DataSource;
+import javax.sql.XAConnection;
+import javax.sql.XADataSource;
+
+import io.helidon.integrations.jdbc.AbstractDataSource;
+
+import jakarta.transaction.TransactionSynchronizationRegistry;
+
+/**
+ * An {@link AbstractDataSource} that wraps another {@link DataSource} that might not behave correctly in the presence
+ * of JTA transaction management, such as one supplied by any of several freely and commercially available connection
+ * pools, and that makes such a non-JTA-aware {@link DataSource} behave as sensibly as possible in the presence of a
+ * JTA-managed transaction.
+ */
+public final class JtaAdaptingDataSource extends AbstractDataSource {
+
+
+    /*
+     * Instance fields.
+     */
+
+
+    private final AuthenticatedConnectionSupplier acs;
+
+    private final UnauthenticatedConnectionSupplier uacs;
+
+
+    /*
+     * Constructors.
+     */
+
+
+    /**
+     * Creates a new {@link JtaAdaptingDataSource} that wraps the supplied {@link DataSource} and helps its {@linkplain
+     * DataSource#getConnection() connections} participate in XA transactions.
+     *
+     * <p>Behavior is left deliberately undefined if the supplied {@link DataSource}'s {@link
+     * DataSource#getConnection()} or {@link DataSource#getConnection(String, String)} methods are implemented to return
+     * or augment the return value of an invocation of the {@link javax.sql.PooledConnection#getConnection()
+     * XAConnection#getConnection()} method.  Less formally, and in general, this class is deliberately not designed to
+     * work with JDBC constructs that are already XA-aware.</p>
+     *
+     * @param ts a {@link TransactionSupplier}; must not be {@code null}
+     *
+     * @param tsr a {@link TransactionSynchronizationRegistry}; must not be {@code null}
+     *
+     * @param interposedSynchronizations whether any {@link jakarta.transaction.Synchronization Synchronization}s
+     * registered should be registered as interposed synchronizations; see {@link
+     * TransactionSynchronizationRegistry#registerInterposedSynchronization(jakarta.transaction.Synchronization)} and
+     * {@link jakarta.transaction.Transaction#registerSynchronization(jakarta.transaction.Synchronization)}
+     *
+     * @param ec an {@link ExceptionConverter}; may be {@code null} in which case a default implementation will be used
+     * instead
+     *
+     * @param ds a {@link DataSource} that may not be XA-compliant; must not be {@code null}; normally supplied by a
+     * connection pool implementation
+     *
+     * @param immediateEnlistment whether attempts to enlist new {@link Connection}s in a global transaction should be
+     * made immediately upon {@link Connection} allocation
+     *
+     * @exception NullPointerException if {@code ts}, {@code tsr} or {@code ds} is {@code null}
+     */
+    public JtaAdaptingDataSource(TransactionSupplier ts,
+                                 TransactionSynchronizationRegistry tsr,
+                                 boolean interposedSynchronizations,
+                                 ExceptionConverter ec,
+                                 DataSource ds,
+                                 boolean immediateEnlistment) {
+        super();
+        Objects.requireNonNull(ts, "ts");
+        Objects.requireNonNull(tsr, "tsr");
+        if (ds instanceof XADataSource xads) {
+            // Some connection pools supply XADataSource objects that pool XAConnections while also unconventionally
+            // exposing them to end users, and also implement the javax.sql.DataSource interface by throwing exceptions
+            // when its methods are invoked. Although XAConnections are not intended for end users, when logical
+            // representations of pooled XAConnections *are* supplied to end users by such a connection pool, those
+            // "borrowed" representations must be "returned" via their close() methods, counter to the documentation of
+            // the (inherited) PooledConnection#close() method, which reads, in part: "An application never calls this
+            // method directly; it is called by the connection pool module, or manager."  As of this writing this branch
+            // of this constructor implements this non-standard behavior.
+            this.acs =
+                (u, p) -> xa(ts, tsr, interposedSynchronizations, ec, xads.getXAConnection(u, p), immediateEnlistment, true);
+            this.uacs = () -> xa(ts, tsr, interposedSynchronizations, ec, xads.getXAConnection(), immediateEnlistment, true);
+        } else {
+            Objects.requireNonNull(ds, "ds");
+            this.acs =
+                (u, p) -> new JtaConnection(ts, tsr, interposedSynchronizations, ec, ds.getConnection(u, p), immediateEnlistment);
+            this.uacs = () -> new JtaConnection(ts, tsr, interposedSynchronizations, ec, ds.getConnection(), immediateEnlistment);
+        }
+    }
+
+    /**
+     * Creates a new {@link JtaAdaptingDataSource} that adapts the supplied {@link XADataSource} and helps {@link
+     * Connection}s it indirectly supplies (by way of its {@linkplain XADataSource#getXAConnection() associated
+     * <code>XAConnection</code>}) participate in XA transactions.
+     *
+     * @param ts a {@link TransactionSupplier}; must not be {@code null}
+     *
+     * @param tsr a {@link TransactionSynchronizationRegistry}; must not be {@code null}
+     *
+     * @param interposedSynchronizations whether any {@link jakarta.transaction.Synchronization Synchronization}s
+     * registered should be registered as interposed synchronizations; see {@link
+     * TransactionSynchronizationRegistry#registerInterposedSynchronization(jakarta.transaction.Synchronization)} and
+     * {@link jakarta.transaction.Transaction#registerSynchronization(jakarta.transaction.Synchronization)}
+     *
+     * @param ec an {@link ExceptionConverter}; may be {@code null} in which case a default implementation will be used
+     * instead
+     *
+     * @param xads an {@link XADataSource} supplied by a connection pool implementation; must not be {@code null}
+     *
+     * @param immediateEnlistment whether attempts to enlist new {@link Connection}s in a global transaction should be
+     * made immediately upon {@link Connection} allocation
+     *
+     * @param closeXac whether or not {@link XAConnection}s {@linkplain XADataSource#getXAConnection() supplied} by the
+     * supplied {@link XADataSource} should be {@linkplain javax.sql.PooledConnection#close() closed} when {@linkplain
+     * XAConnection#getConnection() their <code>Connection</code>}s are {@linkplain Connection#close() closed} (in a
+     * non-standard fashion)
+     *
+     * @exception NullPointerException if {@code ts}, {@code tsr} or {@code xads} is {@code null}
+     *
+     * @deprecated This constructor exists only to handle certain XA-aware connection pools that allow an end-user
+     * caller to "borrow" {@link XAConnection}s and to "return" them using their {@link
+     * javax.sql.PooledConnection#close() close()} methods, a non-standard practice which is discouraged by the
+     * documentation of {@link javax.sql.PooledConnection} (from which {@link XAConnection} inherits).  For
+     * such connection pools, {@link XAConnection}s that are "borrowed" must be returned in this manner to avoid leaks.
+     * This constructor implements this behavior.  Before using it, you should make sure that the connection pool in
+     * question implementing or supplying the {@link XADataSource} has the behavior described above; normally an {@link
+     * XAConnection} should not be used directly or closed by end-user code.
+     */
+    @Deprecated(since = "3.1.0")
+    public JtaAdaptingDataSource(TransactionSupplier ts,
+                                 TransactionSynchronizationRegistry tsr,
+                                 boolean interposedSynchronizations,
+                                 ExceptionConverter ec,
+                                 XADataSource xads,
+                                 boolean immediateEnlistment,
+                                 boolean closeXac) {
+        super();
+        Objects.requireNonNull(xads, "xads");
+        Objects.requireNonNull(ts, "ts");
+        Objects.requireNonNull(tsr, "tsr");
+        // Some connection pools supply XADataSource objects that pool XAConnections. In all likelihood, they should not
+        // do this, given that XAConnections are supposed to be for the innards of connection pools, not for end users.
+        // Nevertheless, when such XAConnections are pooled and logical representations of them are supplied by such a
+        // connection pool, those representations must be closed, counter to the documentation of the (inherited)
+        // PooledConnection#close() method, which reads, in part: "An application never calls this method directly; it
+        // is called by the connection pool module, or manager."  As of this writing this constructor permits this
+        // non-standard behavior when closeXac is true.
+        this.acs =
+            (u, p) -> xa(ts, tsr, interposedSynchronizations, ec, xads.getXAConnection(u, p), immediateEnlistment, closeXac);
+        this.uacs = () -> xa(ts, tsr, interposedSynchronizations, ec, xads.getXAConnection(), immediateEnlistment, closeXac);
+    }
+
+    @Override // DataSource
+    public Connection getConnection(String username, String password) throws SQLException {
+        return this.acs.getConnection(username, password);
+    }
+
+    @Override // DataSource
+    public Connection getConnection() throws SQLException {
+        return this.uacs.getConnection();
+    }
+
+
+    /*
+     * Static methods.
+     */
+
+
+    @Deprecated
+    private static JtaConnection xa(TransactionSupplier ts,
+                                    TransactionSynchronizationRegistry tsr,
+                                    boolean interposedSynchronizations,
+                                    ExceptionConverter ec,
+                                    XAConnection xac,
+                                    boolean immediateEnlistment,
+                                    boolean closeXac)
+        throws SQLException {
+        if (closeXac) {
+            // Some connection pools allow you to "borrow" XAConnections. XAConnections were never intended to be
+            // exposed to end users in this fashion. To return a "borrowed" XAConnection, you invoke its close() method,
+            // which violates the contract of PooledConnection#close(), which XAConnection inherits, whose documentation
+            // reads: "An application never calls this method directly; it is called by the connection pool module, or
+            // manager." This branch of this method implements this non-standard behavior, ensuring that both the
+            // Connection and its sourcing XAConnection are closed appropriately.
+            return new JtaConnection(ts,
+                                     tsr,
+                                     interposedSynchronizations,
+                                     ec,
+                                     xac.getConnection(),
+                                     xac::getXAResource,
+                                     immediateEnlistment) {
+                @Override
+                protected void onClose() throws SQLException {
+                    xac.close();
+                }
+            };
+        }
+        return
+            new JtaConnection(ts,
+                              tsr,
+                              interposedSynchronizations,
+                              ec,
+                              xac.getConnection(),
+                              xac::getXAResource,
+                              immediateEnlistment);
+    }
+
+
+    /*
+     * Inner and nested classes.
+     */
+
+
+    @FunctionalInterface
+    private interface UnauthenticatedConnectionSupplier {
+
+        Connection getConnection() throws SQLException;
+
+    }
+
+    @FunctionalInterface
+    private interface AuthenticatedConnectionSupplier {
+
+        Connection getConnection(String username, String password) throws SQLException;
+
+    }
+
+}
diff --git a/integrations/jta/jdbc/src/main/java/io/helidon/integrations/jta/jdbc/JtaConnection.java b/integrations/jta/jdbc/src/main/java/io/helidon/integrations/jta/jdbc/JtaConnection.java
new file mode 100644
index 00000000000..63146d85348
--- /dev/null
+++ b/integrations/jta/jdbc/src/main/java/io/helidon/integrations/jta/jdbc/JtaConnection.java
@@ -0,0 +1,1214 @@
+/*
+ * Copyright (c) 2023 Oracle and/or its affiliates.
+ *
+ * Licensed 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 io.helidon.integrations.jta.jdbc;
+
+import java.lang.invoke.MethodHandles;
+import java.lang.invoke.VarHandle;
+import java.sql.Array;
+import java.sql.Blob;
+import java.sql.CallableStatement;
+import java.sql.Clob;
+import java.sql.Connection;
+import java.sql.DatabaseMetaData;
+import java.sql.NClob;
+import java.sql.PreparedStatement;
+import java.sql.SQLClientInfoException;
+import java.sql.SQLException;
+import java.sql.SQLNonTransientConnectionException;
+import java.sql.SQLNonTransientException;
+import java.sql.SQLTransientException;
+import java.sql.SQLXML;
+import java.sql.Savepoint;
+import java.sql.ShardingKey;
+import java.sql.Statement;
+import java.sql.Struct;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Properties;
+import java.util.concurrent.Executor;
+import java.util.function.Consumer;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import javax.transaction.xa.XAResource;
+import javax.transaction.xa.Xid;
+
+import io.helidon.integrations.jdbc.ConditionallyCloseableConnection;
+import io.helidon.integrations.jdbc.SQLSupplier;
+import io.helidon.integrations.jdbc.UncheckedSQLException;
+
+import jakarta.transaction.RollbackException;
+import jakarta.transaction.Status;
+import jakarta.transaction.Synchronization;
+import jakarta.transaction.SystemException;
+import jakarta.transaction.Transaction;
+import jakarta.transaction.TransactionSynchronizationRegistry;
+
+import static javax.transaction.xa.XAResource.TMSUCCESS;
+
+/**
+ * A JDBC 4.3-compliant {@link ConditionallyCloseableConnection} that can participate in a {@link Transaction}.
+ */
+class JtaConnection extends ConditionallyCloseableConnection {
+
+
+    /*
+     * Static fields.
+     */
+
+
+    private static final Logger LOGGER = Logger.getLogger(JtaConnection.class.getName());
+
+    // The standard SQL state used for unspecified connection exceptions. Used in this class primarily to indicate
+    // premature connection closure.
+    private static final String CONNECTION_EXCEPTION_NO_SUBCLASS = "08000";
+
+    // The standard SQL state used for transaction-related issues.
+    private static final String INVALID_TRANSACTION_STATE_NO_SUBCLASS = "25000";
+
+    // IBM's proprietary but very descriptive, useful and specific SQL state for when a savepoint operation has been
+    // attempted during a global transaction ("A SAVEPOINT, RELEASE SAVEPOINT, or ROLLBACK TO SAVEPOINT is not allowed
+    // in a trigger, function, or global transaction").
+    private static final String PROHIBITED_SAVEPOINT_OPERATION = "3B503";
+
+    // SQL state 40000 ("transaction rollback, no subclass") is unclear whether it means the SQL/local transaction or
+    // the XA branch transaction or both. It's a convenient SQLState to use nonetheless for handling converting
+    // RollbackExceptions to SQLExceptions.
+    private static final String TRANSACTION_ROLLBACK = "40000";
+
+    private static final VarHandle ENLISTMENT;
+
+    static {
+        try {
+            ENLISTMENT = MethodHandles.lookup().findVarHandle(JtaConnection.class, "enlistment", Enlistment.class);
+        } catch (IllegalAccessException | NoSuchFieldException e) {
+            throw (ExceptionInInitializerError) new ExceptionInInitializerError(e.getMessage()).initCause(e);
+        }
+    }
+
+
+    /*
+     * Instance fields.
+     */
+
+
+    /**
+     * A supplier of {@link Transaction} objects. Often initialized to {@link
+     * jakarta.transaction.TransactionManager#getTransaction() transactionManager::getTransaction}.
+     *
+     * <p>This field is never {@code null}.</p>
+     *
+     * @see TransactionSupplier
+     *
+     * @see jakarta.transaction.TransactionManager#getTransaction()
+     */
+    private final TransactionSupplier ts;
+
+    /**
+     * A {@link TransactionSynchronizationRegistry}.
+     *
+     * <p>This field is never {@code null}.</p>
+     *
+     * @see TransactionSynchronizationRegistry
+     */
+    private final TransactionSynchronizationRegistry tsr;
+
+    /**
+     * Whether any {@link Synchronization}s registered by this {@link JtaConnection}
+     * should be registered as interposed synchronizations.
+     *
+     * @see TransactionSynchronizationRegistry#registerInterposedSynchronization(Synchronization)
+     *
+     * @see Transaction#registerSynchronization(Synchronization)
+     */
+    private final boolean interposedSynchronizations;
+
+    private final SQLSupplier<? extends XAResource> xaResourceSupplier;
+
+    private final Consumer<? super Xid> xidConsumer;
+
+    private volatile Enlistment enlistment;
+
+
+    /*
+     * Constructors.
+     */
+
+
+    /**
+     * Creates a new {@link JtaConnection}.
+     *
+     * @param transactionSupplier a {@link TransactionSupplier}; must not be {@code null}; often {@link
+     * jakarta.transaction.TransactionManager#getTransaction() transactionManager::getTransaction}
+     *
+     * @param transactionSynchronizationRegistry a {@link TransactionSynchronizationRegistry}; must not be {@code null}
+     *
+     * @param interposedSynchronizations whether any {@link Synchronization}s registered by this {@link JtaConnection}
+     * should be registered as interposed synchronizations; see {@link
+     * TransactionSynchronizationRegistry#registerInterposedSynchronization(Synchronization)} and {@link
+     * Transaction#registerSynchronization(Synchronization)}
+     *
+     * @param exceptionConverter an {@link ExceptionConverter}; may be {@code null}
+     *
+     * @param delegate a {@link Connection} that was not sourced from an invocation of {@link
+     * javax.sql.XAConnection#getConnection()}; must not be {@code null}
+     *
+     * @param immediateEnlistment whether an attempt to enlist the new {@link JtaConnection} in a global transaction, if
+     * there is one, will be made immediately
+     *
+     * @param exceptionConverter an {@link ExceptionConverter}; may be {@code null}
+     *
+     * @exception SQLException if transaction enlistment fails or the supplied {@code delegate} {@linkplain
+     * Connection#isClosed() is closed}
+     *
+     * @see #JtaConnection(TransactionSupplier, TransactionSynchronizationRegistry, boolean, ExceptionConverter,
+     * Connection, Supplier, Consumer, boolean)
+     */
+    JtaConnection(TransactionSupplier transactionSupplier,
+                  TransactionSynchronizationRegistry transactionSynchronizationRegistry,
+                  boolean interposedSynchronizations,
+                  ExceptionConverter exceptionConverter,
+                  Connection delegate,
+                  boolean immediateEnlistment)
+        throws SQLException {
+        this(transactionSupplier,
+             transactionSynchronizationRegistry,
+             interposedSynchronizations,
+             exceptionConverter,
+             delegate,
+             null,
+             null,
+             immediateEnlistment);
+    }
+
+    /**
+     * Creates a new {@link JtaConnection}.
+     *
+     * @param transactionSupplier a {@link TransactionSupplier}; must not be {@code null}; often {@link
+     * jakarta.transaction.TransactionManager#getTransaction() transactionManager::getTransaction}
+     *
+     * @param transactionSynchronizationRegistry a {@link TransactionSynchronizationRegistry}; must not be {@code null}
+     *
+     * @param interposedSynchronizations whether any {@link Synchronization}s registered by this {@link JtaConnection}
+     * should be registered as interposed synchronizations; see {@link
+     * TransactionSynchronizationRegistry#registerInterposedSynchronization(Synchronization)} and {@link
+     * Transaction#registerSynchronization(Synchronization)}
+     *
+     * @param exceptionConverter an {@link ExceptionConverter}; may be {@code null}; ignored if {@code
+     * xaResourceSupplier} is non-{@code null}
+     *
+     * @param delegate a {@link Connection} that was not sourced from an invocation of {@link
+     * javax.sql.XAConnection#getConnection()}; must not be {@code null}
+     *
+     * @param xaResourceSupplier a {@link SQLSupplier} of an {@link XAResource} to represent this {@link JtaConnection};
+     * may be and often is {@code null} in which case a new {@link LocalXAResource} will be used instead
+     *
+     * @param immediateEnlistment whether an attempt to enlist the new {@link JtaConnection} in a global transaction, if
+     * there is one, will be made immediately
+     *
+     * @exception NullPointerException if {@code transactionSupplier} or {@code transactionSynchronizationRegistry} is
+     * {@code null}
+     *
+     * @exception SQLException if transaction enlistment fails or the supplied {@code delegate} {@linkplain
+     * Connection#isClosed() is closed}
+     *
+     * @see #JtaConnection(TransactionSupplier, TransactionSynchronizationRegistry, boolean, ExceptionConverter,
+     * Connection, Supplier, Consumer, boolean)
+     */
+    JtaConnection(TransactionSupplier transactionSupplier,
+                  TransactionSynchronizationRegistry transactionSynchronizationRegistry,
+                  boolean interposedSynchronizations,
+                  ExceptionConverter exceptionConverter,
+                  Connection delegate,
+                  SQLSupplier<? extends XAResource> xaResourceSupplier,
+                  boolean immediateEnlistment)
+        throws SQLException {
+        this(transactionSupplier,
+             transactionSynchronizationRegistry,
+             interposedSynchronizations,
+             exceptionConverter,
+             delegate,
+             xaResourceSupplier,
+             null,
+             immediateEnlistment);
+    }
+
+    /**
+     * Creates a new {@link JtaConnection}.
+     *
+     * @param transactionSupplier a {@link TransactionSupplier}; must not be {@code null}; often {@link
+     * jakarta.transaction.TransactionManager#getTransaction() transactionManager::getTransaction}
+     *
+     * @param transactionSynchronizationRegistry a {@link TransactionSynchronizationRegistry}; must not be {@code null}
+     *
+     * @param interposedSynchronizations whether any {@link Synchronization}s registered by this {@link JtaConnection}
+     * should be registered as interposed synchronizations; see {@link
+     * TransactionSynchronizationRegistry#registerInterposedSynchronization(Synchronization)} and {@link
+     * Transaction#registerSynchronization(Synchronization)}
+     *
+     * @param exceptionConverter an {@link ExceptionConverter}; may be {@code null}; ignored if {@code
+     * xaResourceSupplier} is non-{@code null}
+     *
+     * @param delegate a {@link Connection} that was not sourced from an invocation of {@link
+     * javax.sql.XAConnection#getConnection()}; must not be {@code null}
+     *
+     * @param xaResourceSupplier a {@link SQLSupplier} of an {@link XAResource} to represent this {@link JtaConnection};
+     * may be and often is {@code null} in which case a new {@link LocalXAResource} will be used instead
+     *
+     * @param xidConsumer a {@link Consumer} of {@link Xid}s that will be invoked when {@code xaResource} is {@code
+     * null} and a new {@link LocalXAResource} has been created and enlisted; may be {@code null}; useful mainly for
+     * testing
+     *
+     * @param immediateEnlistment whether an attempt to enlist the new {@link JtaConnection} in a global transaction, if
+     * there is one, will be made immediately
+     *
+     * @exception NullPointerException if {@code transactionSupplier} or {@code transactionSynchronizationRegistry} is
+     * {@code null}
+     *
+     * @exception SQLException if transaction enlistment fails or the supplied {@code delegate} {@linkplain
+     * Connection#isClosed() is closed}
+     */
+    JtaConnection(TransactionSupplier transactionSupplier,
+                  TransactionSynchronizationRegistry transactionSynchronizationRegistry,
+                  boolean interposedSynchronizations,
+                  ExceptionConverter exceptionConverter,
+                  Connection delegate,
+                  SQLSupplier<? extends XAResource> xaResourceSupplier,
+                  Consumer<? super Xid> xidConsumer,
+                  boolean immediateEnlistment)
+        throws SQLException {
+        super(delegate,
+              true, // closeable
+              true); // strict isClosed checking; always a good thing
+        this.ts = Objects.requireNonNull(transactionSupplier, "transactionSupplier");
+        this.tsr = Objects.requireNonNull(transactionSynchronizationRegistry, "transactionSynchronizationRegistry");
+        if (delegate.isClosed()) {
+            throw new SQLNonTransientConnectionException("delegate is closed", CONNECTION_EXCEPTION_NO_SUBCLASS);
+        }
+        this.interposedSynchronizations = interposedSynchronizations;
+        this.xaResourceSupplier =
+            xaResourceSupplier == null
+            ? () -> new LocalXAResource(this::connectionFunction, exceptionConverter)
+            : xaResourceSupplier;
+        this.xidConsumer = xidConsumer == null ? JtaConnection::sink : xidConsumer;
+        if (immediateEnlistment) {
+            this.enlist();
+        }
+    }
+
+
+    /*
+     * Instance methods.
+     */
+
+
+    @Override // ConditionallyCloseableConnection
+    public final void setCloseable(boolean closeable) {
+        super.setCloseable(closeable);
+        if (LOGGER.isLoggable(Level.FINER)) {
+            LOGGER.entering(this.getClass().getName(), "setCloseable", closeable);
+            LOGGER.exiting(this.getClass().getName(), "setCloseable");
+        }
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final Statement createStatement() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.createStatement();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final PreparedStatement prepareStatement(String sql) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.prepareStatement(sql);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final CallableStatement prepareCall(String sql) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.prepareCall(sql);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final String nativeSQL(String sql) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.nativeSQL(sql);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void setAutoCommit(boolean autoCommit) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        if (autoCommit && this.enlisted()) {
+            // "SQLException...if...setAutoCommit(true) is called while participating in a distributed transaction"
+            throw new SQLNonTransientException("Connection enlisted in transaction", INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+        }
+        super.setAutoCommit(autoCommit);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final boolean getAutoCommit() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.getAutoCommit();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void commit() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        if (this.enlisted()) {
+            // "SQLException...if...this method is called while participating in a distributed transaction"
+            throw new SQLNonTransientException("Connection enlisted in transaction", INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+        }
+        super.commit();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void rollback() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        if (this.enlisted()) {
+            // "SQLException...if...this method is called while participating in a distributed transaction"
+            throw new SQLNonTransientException("Connection enlisted in transaction", INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+        }
+        super.rollback();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final DatabaseMetaData getMetaData() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.getMetaData();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void setReadOnly(boolean readOnly) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        super.setReadOnly(readOnly);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final boolean isReadOnly() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.isReadOnly();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void setCatalog(String catalog) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        super.setCatalog(catalog);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final String getCatalog() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.getCatalog();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void setTransactionIsolation(int level) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        super.setTransactionIsolation(level);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final int getTransactionIsolation() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.getTransactionIsolation();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final Statement createStatement(int resultSetType, int resultSetConcurrency) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.createStatement(resultSetType, resultSetConcurrency);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final PreparedStatement prepareStatement(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.prepareStatement(sql, resultSetType, resultSetConcurrency);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.prepareCall(sql, resultSetType, resultSetConcurrency);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final Map<String, Class<?>> getTypeMap() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.getTypeMap();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void setTypeMap(Map<String, Class<?>> map) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        super.setTypeMap(map);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void setHoldability(int holdability) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        super.setHoldability(holdability);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final int getHoldability() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.getHoldability();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final Savepoint setSavepoint() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        if (this.enlisted()) {
+            // "SQLException...if...this method is called while participating in a distributed transaction"
+            throw new SQLNonTransientException("Connection enlisted in transaction", PROHIBITED_SAVEPOINT_OPERATION);
+        }
+        return super.setSavepoint();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final Savepoint setSavepoint(String name) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        if (this.enlisted()) {
+            // "SQLException...if...this method is called while participating in a distributed transaction"
+            throw new SQLNonTransientException("Connection enlisted in transaction", PROHIBITED_SAVEPOINT_OPERATION);
+        }
+        return super.setSavepoint(name);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void rollback(Savepoint savepoint) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        if (this.enlisted()) {
+            // "SQLException...if...this method is called while participating in a distributed transaction"
+            throw new SQLNonTransientException("Connection enlisted in transaction", PROHIBITED_SAVEPOINT_OPERATION);
+        }
+        super.rollback(savepoint);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void releaseSavepoint(Savepoint savepoint) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        if (this.enlisted()) {
+            // "SQLException...if...the given Savepoint object is not a valid savepoint in the current transaction"
+            //
+            // Interestingly JDBC doesn't mandate an exception being thrown here if the connection is enlisted in a
+            // global transaction, but it looks like a SQL state such as 3B503 is often thrown in this case.
+            throw new SQLNonTransientException("Connection enlisted in transaction", PROHIBITED_SAVEPOINT_OPERATION);
+        }
+        super.releaseSavepoint(savepoint);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final Statement createStatement(int resultSetType, int resultSetConcurrency, int resultSetHoldability)
+        throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.createStatement(resultSetType, resultSetConcurrency, resultSetHoldability);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final PreparedStatement prepareStatement(String sql, int rsType, int rsConcurrency, int rsHoldability)
+      throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.prepareStatement(sql, rsType, rsConcurrency, rsHoldability);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final CallableStatement prepareCall(String sql, int resultSetType, int resultSetConcurrency, int resultSetHoldability)
+        throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.prepareCall(sql, resultSetType, resultSetConcurrency, resultSetHoldability);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final PreparedStatement prepareStatement(String sql, int autoGeneratedKeys) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.prepareStatement(sql, autoGeneratedKeys);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final PreparedStatement prepareStatement(String sql, int[] columnIndexes) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.prepareStatement(sql, columnIndexes);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final PreparedStatement prepareStatement(String sql, String[] columnNames) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.prepareStatement(sql, columnNames);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final Clob createClob() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.createClob();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final Blob createBlob() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.createBlob();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final NClob createNClob() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.createNClob();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final SQLXML createSQLXML() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.createSQLXML();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final boolean isValid(int timeout) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.isValid(timeout);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void setClientInfo(String name, String value) throws SQLClientInfoException {
+        try {
+            this.failWhenClosed();
+            this.enlist();
+            super.setClientInfo(name, value);
+        } catch (SQLClientInfoException e) {
+            throw e;
+        } catch (SQLException e) {
+            throw new SQLClientInfoException(e.getMessage(), e.getSQLState(), e.getErrorCode(), Map.of(), e);
+        }
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void setClientInfo(Properties properties) throws SQLClientInfoException {
+        try {
+            this.failWhenClosed();
+            this.enlist();
+            super.setClientInfo(properties);
+        } catch (SQLClientInfoException e) {
+            throw e;
+        } catch (SQLException e) {
+            throw new SQLClientInfoException(e.getMessage(), e.getSQLState(), e.getErrorCode(), Map.of(), e);
+        }
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final String getClientInfo(String name) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.getClientInfo(name);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final Properties getClientInfo() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.getClientInfo();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final Array createArrayOf(String typeName, Object[] elements) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.createArrayOf(typeName, elements);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final Struct createStruct(String typeName, Object[] attributes) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.createStruct(typeName, attributes);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void setSchema(String schema) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        super.setSchema(schema);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final String getSchema() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.getSchema();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void abort(Executor executor) throws SQLException {
+        // this.enlist(); // Deliberately omitted, but not by spec.
+
+        // NOTE
+        //
+        // abort(Executor) is a method that seems to be designed for an administrator, and so even if there is a
+        // transaction in progress we probably should allow closing.
+        //
+        // TO DO: should we heuristically roll back? Purge the Xid?
+        this.setCloseable(true);
+        super.abort(executor);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void setNetworkTimeout(Executor executor, int milliseconds) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        super.setNetworkTimeout(executor, milliseconds);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final int getNetworkTimeout() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.getNetworkTimeout();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void beginRequest() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        super.beginRequest();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void endRequest() throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        super.endRequest();
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final boolean setShardingKeyIfValid(ShardingKey shardingKey, ShardingKey superShardingKey, int timeout)
+        throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.setShardingKeyIfValid(shardingKey, superShardingKey, timeout);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final boolean setShardingKeyIfValid(ShardingKey shardingKey, int timeout) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        return super.setShardingKeyIfValid(shardingKey, timeout);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void setShardingKey(ShardingKey shardingKey, ShardingKey superShardingKey) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        super.setShardingKey(shardingKey, superShardingKey);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void setShardingKey(ShardingKey shardingKey) throws SQLException {
+        this.failWhenClosed();
+        this.enlist();
+        super.setShardingKey(shardingKey);
+    }
+
+    @Override // ConditionallyCloseableConnection
+    public final void close() throws SQLException {
+        if (LOGGER.isLoggable(Level.FINER)) {
+            LOGGER.entering(this.getClass().getName(), "close");
+        }
+        // The JTA Specification, section 4.2, has a non-normative diagram illustrating that close() is expected,
+        // but not required, to call Transaction#delistResource(XAResource). This is, mind you, before the
+        // prepare/commit completion cycle has started.
+        Enlistment enlistment = this.enlistment; // volatile read
+        if (enlistment != null) {
+            try {
+                // TMSUCCESS because it's an ordinary close() call, not a delisting due to an exception
+                boolean delisted = enlistment.transaction().delistResource(enlistment.xaResource(), TMSUCCESS);
+                if (LOGGER.isLoggable(Level.FINE)) {
+                    LOGGER.logp(Level.FINE,
+                                this.getClass().getName(), "close",
+                                "{0} {1} from {2}",
+                                new Object[] {delisted ? "Delisted" : "Failed to delist",
+                                              enlistment.xaResource(),
+                                              enlistment.transaction()});
+                }
+            } catch (IllegalStateException e) {
+                // Transaction went from active or marked for rollback to some other state; whatever; we're no longer
+                // enlisted so we didn't delist.
+                if (LOGGER.isLoggable(Level.FINE)) {
+                    LOGGER.logp(Level.FINE, this.getClass().getName(), "close", e.getMessage(), e);
+                }
+            } catch (SystemException e) {
+                throw new SQLTransientException(e.getMessage(), INVALID_TRANSACTION_STATE_NO_SUBCLASS, e);
+            }
+        }
+        super.close();
+        if (LOGGER.isLoggable(Level.FINE) && !this.isClosePending()) {
+            // If a close is not pending then that means it actually happened.
+            LOGGER.logp(Level.FINE, this.getClass().getName(), "close",
+                        "Closed {0} on thread {1}", new Object[] {this, Thread.currentThread()});
+        }
+        if (LOGGER.isLoggable(Level.FINER)) {
+            LOGGER.exiting(this.getClass().getName(), "close");
+        }
+    }
+
+    @Override // Object
+    public final int hashCode() {
+        return System.identityHashCode(this);
+    }
+
+    @Override // Object
+    public final boolean equals(Object other) {
+        return this == other;
+    }
+
+    /**
+     * Returns {@code true} if a JTA transaction exists and {@linkplain
+     * TransactionSynchronizationRegistry#getTransactionStatus() has a status} equal to either {@link
+     * Status#STATUS_ACTIVE} or {@link Status#STATUS_MARKED_ROLLBACK}.
+     *
+     * @return {@code true} if a JTA transaction exists and {@linkplain
+     * TransactionSynchronizationRegistry#getTransactionStatus() has a status} equal to either {@link
+     * Status#STATUS_ACTIVE} or {@link Status#STATUS_MARKED_ROLLBACK}; {@code false} in all other cases
+     *
+     * @exception SQLException if the status could not be acquired
+     *
+     * @see TransactionSynchronizationRegistry#getTransactionStatus()
+     *
+     * @see Status
+     */
+    private boolean activeOrMarkedRollbackTransaction() throws SQLException {
+        switch (this.transactionStatus()) {
+            // See https://www.eclipse.org/lists/jta-dev/msg00264.html and
+            // https://github.com/jakartaee/transactions/issues/211.
+        case Status.STATUS_ACTIVE:
+        case Status.STATUS_MARKED_ROLLBACK:
+            return true;
+        default:
+            return false;
+        }
+    }
+
+    private int transactionStatus() throws SQLException {
+        try {
+            return this.tsr.getTransactionStatus();
+        } catch (RuntimeException e) {
+            // See
+            // https://github.com/jbosstm/narayana/blob/c5f02d07edb34964b64341974ab689ea44536603/ArjunaJTA/jta/classes/com/arjuna/ats/internal/jta/transaction/arjunacore/TransactionSynchronizationRegistryImple.java#L153-L164;
+            // despite it not being documented, TCK-passing implementations of
+            // TransactionSynchronizationRegistry#getTransactionStatus() can apparently throw RuntimeException. Since
+            // getTransactionStatus() is specified to return "the result of executing TransactionManager.getStatus() in
+            // the context of the transaction bound to the current thread at the time this method is called", it follows
+            // that possible SystemExceptions thrown by TransactionManager#getStatus() implementations will have to be
+            // dealt with in *some* way, even though the javadoc for
+            // TransactionSynchronizationRegistry#getTransactionStatus() does not account for such a thing.
+            throw new SQLTransientException(e.getMessage(), INVALID_TRANSACTION_STATE_NO_SUBCLASS, e);
+        }
+    }
+
+    /**
+     * Returns {@code true} if and only if this {@link JtaConnection} is associated with a JTA transaction whose
+     * {@linkplain Transaction#getStatus() status} is one of {@link Status#STATUS_ACTIVE} or {@link
+     * Status#STATUS_MARKED_ROLLBACK} as a result of a prior {@link #enlist()} invocation on the current thread.
+     *
+     * @return {@code true} if and only if this {@link JtaConnection} is associated with a {@linkplain
+     * #activeOrMarkedRollbackTransaction() JTA transaction whose status is known and not yet prepared}; {@code false}
+     * in all other cases
+     *
+     * @exception SQLException if the enlisted status could not be acquired
+     */
+    boolean enlisted() throws SQLException {
+        Enlistment enlistment = this.enlistment; // volatile read
+        if (enlistment == null) {
+            return false;
+        } else if (enlistment.threadId() != Thread.currentThread().getId()) {
+            // We're enlisted in a Transaction on thread 1, and a caller from thread 2 is trying to do something with
+            // us. This could have unintended side effects. Throw.
+            throw new SQLTransientException("Already enlisted (" + enlistment + "); current thread id: "
+                                            + Thread.currentThread().getId(), INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+        }
+
+        // We are, or were, enlisted. Let's see in what way.
+
+        // We're enlisted in a Transaction that was created on the current thread. So far so good. Is it active?
+        Transaction t = enlistment.transaction();
+        int transactionStatus = statusFrom(t);
+        switch (transactionStatus) {
+        case Status.STATUS_ACTIVE:
+            // We have been enlisted in an active transaction that was created on this thread. Is the current
+            // transaction, whatever it is, active?
+            int currentThreadTransactionStatus = this.transactionStatus();
+            switch (currentThreadTransactionStatus) {
+            case Status.STATUS_ACTIVE:
+                // We have been enlisted in an active transaction that was created on this thread AND the current
+                // thread's transaction status is ALSO active. Is the current thread's transaction equal to the one
+                // we're enlisted in?  Or is it a different one (in which case our transaction has been suspended)?
+                if (t.equals(this.transaction())) {
+                    // The Transaction associated with the current thread is the active one we're enlisted with, so
+                    // we're already enlisted and everything is fine. (Equality is governed by the spec:
+                    // https://jakarta.ee/specifications/transactions/2.0/jakarta-transactions-spec-2.0.html#transaction-equality-and-hash-code)
+                    return true;
+                }
+                throw new SQLTransientException("Attempting to perform work while associated with a suspended transaction",
+                                                INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+            case Status.STATUS_COMMITTED:
+            case Status.STATUS_ROLLEDBACK:
+            case Status.STATUS_COMMITTING:
+            case Status.STATUS_MARKED_ROLLBACK:
+            case Status.STATUS_PREPARED:
+            case Status.STATUS_PREPARING:
+            case Status.STATUS_ROLLING_BACK:
+            case Status.STATUS_NO_TRANSACTION:
+                // The current thread's transaction status is not active, so it can't be the same transaction with
+                // which we are enlisted, so by definition we're enlisted in a suspended transaction and no further
+                // work should be done until that transaction is resumed.
+                throw new SQLTransientException("Attempting to perform work while associated with a suspended transaction",
+                                                INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+            case Status.STATUS_UNKNOWN:
+            default:
+                // Unexpected or illegal. Throw.
+                throw new SQLTransientException("Unexpected transaction status: " + currentThreadTransactionStatus,
+                                                INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+            }
+        case Status.STATUS_COMMITTED:
+        case Status.STATUS_ROLLEDBACK:
+            // We *were* enlisted, and sort of still are (there's a non-null Enlistment), but now the transaction is
+            // irrevocably completed, and the non-null Enlistment will be removed momentarily by another thread
+            // executing our transactionCompleted(int) method, so really for all intents and purposes we're no
+            // longer enlisted.
+            return false;
+        case Status.STATUS_COMMITTING:
+        case Status.STATUS_MARKED_ROLLBACK:
+        case Status.STATUS_PREPARED:
+        case Status.STATUS_PREPARING:
+        case Status.STATUS_ROLLING_BACK:
+            // Interim or effectively interim. Throw to prevent accidental side effects.
+            throw new SQLTransientException("Non-terminal transaction status: " + transactionStatus,
+                                            INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+        case Status.STATUS_NO_TRANSACTION:
+            // Somehow an Enlistment was created with a Transaction in the Status.STATUS_NO_TRANSACTION status. This
+            // should absolutely never happen.
+            throw new AssertionError();
+        case Status.STATUS_UNKNOWN:
+        default:
+            // Unexpected or illegal. Throw.
+            throw new SQLTransientException("Unexpected transaction status: " + transactionStatus,
+                                            INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+        }
+    }
+
+    private Transaction transaction() throws SQLException {
+        try {
+            return this.ts.getTransaction();
+        } catch (RuntimeException | SystemException e) {
+            throw new SQLTransientException(e.getMessage(), INVALID_TRANSACTION_STATE_NO_SUBCLASS, e);
+        }
+    }
+
+    /**
+     * Attempts to enlist this {@link JtaConnection} in the current JTA transaction, if there is one, and its status is
+     * {@link Status#STATUS_ACTIVE}, and this {@link JtaConnection} is not already {@linkplain #enlisted() enlisted}.
+     *
+     * @exception SQLException if a transaction-related error occurs, or if the return value of an invocation of the
+     * {@link #getAutoCommit()} method returns {@code false}
+     *
+     * @see #enlisted()
+     */
+    @SuppressWarnings("checkstyle:MethodLength")
+    private void enlist() throws SQLException {
+
+        if (this.enlisted()) {
+            return;
+        }
+
+        int currentThreadTransactionStatus = this.transactionStatus();
+        switch (currentThreadTransactionStatus) {
+        case Status.STATUS_ACTIVE:
+            // There is a global transaction currently active on the current thread. That's good. Keep going.
+            break;
+        case Status.STATUS_COMMITTED:
+        case Status.STATUS_NO_TRANSACTION:
+        case Status.STATUS_ROLLEDBACK:
+            // There is no global transaction or there is very shortly about to be no global transaction on the current
+            // thread; this is an effectively terminal state. Enlistment is impossible or silly. Very common. Return
+            // without enlisting.
+            return;
+        case Status.STATUS_COMMITTING:
+        case Status.STATUS_MARKED_ROLLBACK:
+        case Status.STATUS_PREPARED:
+        case Status.STATUS_PREPARING:
+        case Status.STATUS_ROLLING_BACK:
+            // Interim or effectively interim state. Uncommon. Throw to prevent accidental side effects.
+            throw new SQLTransientException("Non-terminal current thread transaction status: " + currentThreadTransactionStatus,
+                                            INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+        case Status.STATUS_UNKNOWN:
+        default:
+            // Unexpected or illegal state. Throw.
+            throw new SQLTransientException("Unexpected current thread transaction status: " + currentThreadTransactionStatus,
+                                            INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+        }
+
+        if (!super.getAutoCommit()) {
+            // There is, as far as we can tell, an active global transaction on the current thread, and
+            // super.getAutoCommit() (super. on purpose, not this.) returned false, and we aren't (yet) enlisted with
+            // the active global transaction, so autoCommit must have been disabled on purpose by the caller, not by the
+            // transaction enlistment machinery. In such a case, we don't want to permit enlistment, because a local
+            // transaction may be in progress and we don't want to have its effects mixed in.
+            throw new SQLTransientException("autoCommit was false during active transaction enlistment",
+                                            INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+        }
+
+        Transaction t = this.transaction();
+        int transactionStatus = statusFrom(t);
+        switch (transactionStatus) {
+        case Status.STATUS_ACTIVE:
+            // No one has started or finished the transaction completion process yet. Most common. Keep going.
+            break;
+        case Status.STATUS_COMMITTED:
+        case Status.STATUS_ROLLEDBACK:
+            // Terminal. Return without enlisting.
+            return;
+        case Status.STATUS_COMMITTING:
+        case Status.STATUS_MARKED_ROLLBACK:
+        case Status.STATUS_PREPARED:
+        case Status.STATUS_PREPARING:
+        case Status.STATUS_ROLLING_BACK:
+            // Interim or effectively interim. Throw to prevent accidental side effects.
+            throw new SQLTransientException("Non-terminal transaction status: " + transactionStatus,
+                                            INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+        case Status.STATUS_NO_TRANSACTION:
+            // Impossible state machine transition since t is non-null and at least at one earlier point on this thread
+            // the status was Status.STATUS_ACTIVE. Even if somehow the global transaction is disassociated from the
+            // current thread, the Transaction object's status will never, by spec, go to
+            // Status.STATUS_NO_TRANSACTION. See
+            // https://groups.google.com/g/narayana-users/c/eYVUmhE9QZg/m/xbBh2CsBBQAJ.
+            throw new AssertionError(); // per spec
+        case Status.STATUS_UNKNOWN:
+        default:
+            // Unexpected or illegal. Throw.
+            throw new SQLTransientException("Unexpected transaction status: " + transactionStatus,
+                                            INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+        }
+
+        // One last check to see if we've been actually closed or requested to close asynchronously.
+        this.failWhenClosed();
+
+        // Point of no return. We ensured that the Transaction at one point had a status of Status.STATUS_ACTIVE and
+        // ensured we aren't already enlisted and our autoCommit status is true. The Transaction's status can still
+        // change at any point (as a result of asynchronous rollback, for example) through certain permitted state
+        // transitions, so we have to watch for exceptions.
+
+        XAResource xar = this.xaResourceSupplier.get();
+        if (xar == null) {
+            throw new SQLTransientException("xaResourceSupplier.get() == null");
+        }
+
+        Enlistment enlistment = new Enlistment(t, xar);
+        if (!ENLISTMENT.compareAndSet(this, null, enlistment)) { // atomic volatile write
+            // Setting this.enlistment could conceivably fail if another thread already enlisted this JtaConnection.
+            // That would be bad.
+            //
+            // (The this.enlistment read in the exception message is a volatile read, and thanks to the compareAndSet
+            // call above we know it is non-null.)
+            throw new SQLTransientException("Already enlisted (" + this.enlistment
+                                            + "); current transaction: " + t
+                                            + "; current thread id: " + Thread.currentThread().getId(),
+                                            INVALID_TRANSACTION_STATE_NO_SUBCLASS);
+        }
+
+        try {
+            // The XAResource is placed into the TransactionSynchronizationRegistry so that it can be delisted if
+            // appropriate during invocation of the close() method (q.v). We do it before the actual
+            // Transaction#enlistResource(XAResource) call on purpose.
+            if (this.interposedSynchronizations) {
+                this.tsr.registerInterposedSynchronization((Sync) this::transactionCompleted);
+                if (LOGGER.isLoggable(Level.FINE)) {
+                    LOGGER.logp(Level.FINE, this.getClass().getName(), "connectionFunction",
+                                "Registered interposed synchronization (transactionCompleted(int)) for {0}", t);
+                }
+            } else {
+                t.registerSynchronization((Sync) this::transactionCompleted);
+                if (LOGGER.isLoggable(Level.FINE)) {
+                    LOGGER.logp(Level.FINE, this.getClass().getName(), "connectionFunction",
+                                "Registered synchronization (transactionCompleted(int)) for {0}", t);
+                }
+            }
+            // Don't let our caller close us "for real". close() invocations will be recorded as pending. See
+            // #close(). (Note that this is "undone" after transaction completion in #transactionCompleted(int), which
+            // was just registered as a synchronization immediately above.)
+            this.setCloseable(false);
+            // (Guaranteed to call xar.start(Xid, int) on this thread.)
+            t.enlistResource(xar);
+            if (LOGGER.isLoggable(Level.FINE)) {
+                LOGGER.logp(Level.FINE,
+                            this.getClass().getName(), "enlist",
+                            "Enlisted {0} in transaction {1}", new Object[] {xar, t});
+            }
+        } catch (RollbackException e) {
+            this.enlistment = null; // volatile write
+            // The enlistResource(XAResource) or registerSynchronization(Synchronization) operation failed because the
+            // transaction was rolled back.
+            throw new SQLNonTransientException(e.getMessage(), TRANSACTION_ROLLBACK, e);
+        } catch (RuntimeException | SystemException e) {
+            this.enlistment = null; // volatile write
+            if (e.getCause() instanceof RollbackException) {
+                // The enlistResource(XAResource) operation failed because the transaction was rolled back.
+                throw new SQLNonTransientException(e.getMessage(), TRANSACTION_ROLLBACK, e);
+            }
+            // The t.enlistResource(XAResource) operation failed, or the
+            // tsr.registerInterposedSynchronization(Synchronization) operation failed, or the
+            // t.registerSynchronization(Synchronization) operation failed. In any case, no XAResource was actually
+            // enlisted.
+            throw new SQLTransientException(e.getMessage(), INVALID_TRANSACTION_STATE_NO_SUBCLASS, e);
+        } catch (Error e) {
+            this.enlistment = null; // volatile write
+            throw e;
+        }
+    }
+
+    // (Used only by reference by LocalXAResource#start(Xid, int) as a result of calling
+    // Transaction#enlistResource(XAResource) in enlist() above.)
+    private Connection connectionFunction(Xid xid) {
+        this.xidConsumer.accept(xid);
+        return this.delegate();
+    }
+
+    // (Used only by reference in enlist() above. Remember, this callback may be called by the TransactionManager on
+    // any thread at any time for any reason.)
+    private void transactionCompleted(int commitedOrRolledBack) {
+        this.enlistment = null; // volatile write
+        try {
+            if (!super.isClosed()) {
+                // Did someone call close() while we were enlisted?  That's permitted by section 4.2 of the JTA
+                // specification, but it shouldn't actually close the connection because the prepare/commit cycle
+                // wouldn't have started.
+                boolean closeWasPending = this.isClosePending(); // volatile read
+
+                // If they did, then we must be non-closeable. If they didn't, we could be either closeable or not
+                // closeable.
+                assert !closeWasPending || !this.isCloseable();
+
+                // Now the global transaction is over, so blindly set our closeable status to true. (It may already be
+                // true.)  This resets the closePending status, per spec.
+
+                this.setCloseable(true);
+                assert this.isCloseable();
+                assert !this.isClosePending();
+
+                // If there WAS a close() attempt, now it CAN work, so let it work.
+                if (closeWasPending) {
+                    this.close();
+                }
+            }
+        } catch (SQLException e) {
+            // (Synchronization implementations can throw only unchecked exceptions.)
+            throw new UncheckedSQLException(e);
+        }
+    }
+
+
+    /*
+     * Static methods.
+     */
+
+
+    private static int statusFrom(Transaction t) throws SQLException {
+        Objects.requireNonNull(t, "t");
+        try {
+            return t.getStatus();
+        } catch (RuntimeException | SystemException e) {
+            throw new SQLTransientException(e.getMessage(), INVALID_TRANSACTION_STATE_NO_SUBCLASS, e);
+        }
+    }
+
+    private static void sink(Object ignored) {
+
+    }
+
+
+    /*
+     * Inner and nested classes.
+     */
+
+
+    /**
+     * A functional {@link Synchronization}.
+     *
+     * @see Synchronization
+     */
+    @FunctionalInterface
+    interface Sync extends Synchronization {
+
+        /**
+         * Called prior to the start of the two-phase transaction commit process.
+         *
+         * <p>The default implementation of this method does nothing.</p>
+         *
+         * @see Synchronization#beforeCompletion()
+         */
+        default void beforeCompletion() {
+
+        }
+
+    }
+
+    private static final record Enlistment(long threadId, Transaction transaction, XAResource xaResource) {
+
+        private Enlistment(Transaction transaction, XAResource xaResource) {
+            this(Thread.currentThread().getId(), transaction, xaResource);
+        }
+
+        private Enlistment {
+            Objects.requireNonNull(transaction, "transaction");
+            Objects.requireNonNull(xaResource, "xaResource");
+        }
+
+    }
+
+}
diff --git a/integrations/jta/jdbc/src/test/java/io/helidon/integrations/jta/jdbc/TestJtaConnection.java b/integrations/jta/jdbc/src/test/java/io/helidon/integrations/jta/jdbc/TestJtaConnection.java
new file mode 100644
index 00000000000..3dfa4fe8efd
--- /dev/null
+++ b/integrations/jta/jdbc/src/test/java/io/helidon/integrations/jta/jdbc/TestJtaConnection.java
@@ -0,0 +1,407 @@
+/*
+ * Copyright (c) 2023 Oracle and/or its affiliates.
+ *
+ * Licensed 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 io.helidon.integrations.jta.jdbc;
+
+import java.lang.reflect.Proxy;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.Statement;
+import java.sql.SQLException;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Logger;
+
+import javax.transaction.xa.XAResource;
+import javax.transaction.xa.Xid;
+
+import io.helidon.integrations.jdbc.ConditionallyCloseableConnection;
+import io.helidon.integrations.jta.jdbc.LocalXAResource.Association;
+import io.helidon.integrations.jta.jdbc.LocalXAResource.Association.BranchState;
+
+import jakarta.transaction.HeuristicMixedException;
+import jakarta.transaction.HeuristicRollbackException;
+import jakarta.transaction.InvalidTransactionException;
+import jakarta.transaction.NotSupportedException;
+import jakarta.transaction.RollbackException;
+import jakarta.transaction.Status;
+import jakarta.transaction.Synchronization;
+import jakarta.transaction.SystemException;
+import jakarta.transaction.Transaction;
+import jakarta.transaction.TransactionManager;
+import jakarta.transaction.TransactionSynchronizationRegistry;
+
+import com.arjuna.ats.arjuna.common.Uid;
+import com.arjuna.ats.arjuna.coordinator.TransactionReaper;
+import com.arjuna.ats.arjuna.coordinator.listener.ReaperMonitor;
+import com.arjuna.ats.jta.common.JTAEnvironmentBean;
+import org.h2.jdbcx.JdbcDataSource;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import static io.helidon.integrations.jta.jdbc.LocalXAResource.Association.BranchState.ACTIVE;
+import static io.helidon.integrations.jta.jdbc.LocalXAResource.Association.BranchState.IDLE;
+import static io.helidon.integrations.jta.jdbc.LocalXAResource.ASSOCIATIONS;
+import static jakarta.transaction.Status.STATUS_NO_TRANSACTION;
+import static org.hamcrest.CoreMatchers.instanceOf;
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.not;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.CoreMatchers.nullValue;
+import static org.hamcrest.CoreMatchers.sameInstance;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+final class TestJtaConnection {
+
+    private static final Logger LOGGER = Logger.getLogger(TestJtaConnection.class.getName());
+
+    private static final JTAEnvironmentBean jtaEnvironmentBean = new JTAEnvironmentBean();
+
+    private JdbcDataSource h2ds;
+
+    private TransactionManager tm;
+
+    private TransactionSynchronizationRegistry tsr;
+
+    private TestJtaConnection() throws SQLException {
+        super();
+    }
+
+    @BeforeEach
+    final void initializeH2DataSource() throws SQLException {
+        JdbcDataSource ds = new JdbcDataSource();
+        // Use TRACE_LEVEL_FILE=4 to turn on slf4j logging.
+        ds.setURL("jdbc:h2:mem:test;INIT=SET TRACE_LEVEL_FILE=4");
+        ds.setUser("sa");
+        ds.setPassword("sa");
+        this.h2ds = ds;
+    }
+
+    @BeforeEach
+    final void initializeTransactionManager() throws SystemException {
+        this.tm = jtaEnvironmentBean.getTransactionManager(); // com.arjuna.ats.jta.TransactionManager.transactionManager();
+        this.tm.setTransactionTimeout(20 * 60); // 20 minutes for debugging
+    }
+
+    @BeforeEach
+    final void initializeTransactionSynchronizationRegistry() throws SystemException {
+        this.tsr = jtaEnvironmentBean.getTransactionSynchronizationRegistry();
+    }
+
+    @AfterEach
+    final void rollback() throws SQLException, SystemException {
+        switch (this.tm.getStatus()) {
+        case STATUS_NO_TRANSACTION:
+            break;
+        default:
+            this.tm.rollback();
+            break;
+        }
+        this.tm.setTransactionTimeout(0);
+        // assertThat(ASSOCIATIONS.size(), is(0));
+    }
+
+    @DisplayName("Spike")
+    @SuppressWarnings("try")
+    @Test
+    final void testSpike()
+        throws HeuristicMixedException,
+               HeuristicRollbackException,
+               NotSupportedException,
+               RollbackException,
+               SQLException,
+               SystemException {
+        LOGGER.info("Starting testSpike()");
+        tm.begin();
+
+        Transaction t = tm.getTransaction();
+
+        // For this test, where transaction reaping is set to happen eons in the future, we can make this assertion.
+        // For real-world scenarios, the reaper might have ended the transaction immediately on another thread.
+        assertThat(t.getStatus(), is(Status.STATUS_ACTIVE));
+
+        try (Connection physicalConnection = h2ds.getConnection();
+             JtaConnection logicalConnection = new JtaConnection(tm::getTransaction,
+                                                                 tsr,
+                                                                 true,
+                                                                 null,
+                                                                 physicalConnection,
+                                                                 null,
+                                                                 x -> tsr.putResource("xid", x),
+                                                                 false)) {
+
+            // Make sure everything is hooked up properly.
+            assertThat(logicalConnection.delegate(), sameInstance(physicalConnection));
+
+            // Trigger an Object method; make sure nothing blows up (in case we're using proxies).
+            logicalConnection.toString();
+
+            // Up until this point, the connection should not be enlisted. (delegate() and toString() must not cause
+            // enlistment.)
+            assertThat(logicalConnection.enlisted(), is(false));
+
+            // (Calling enlisted() itself must not cause enlistment.)
+            assertThat(logicalConnection.enlisted(), is(false));
+
+            // Since it's not enlisted, it should be closeable (but it has not yet been closed).
+            assertThat(logicalConnection.isCloseable(), is(true));
+            assertThat(logicalConnection.isClosed(), is(false));
+
+            // Trigger a harmless Connection-related method; make sure nothing blows up.
+            logicalConnection.getHoldability();
+
+            // Almost all Connection methods, including that one, will cause enlistment to happen.
+            assertThat(logicalConnection.enlisted(), is(true));
+
+            // Make sure the XAResource recorded the association.
+            Xid xid = (Xid) this.tsr.getResource("xid");
+            assertThat(ASSOCIATIONS.get(xid).branchState(), is(ACTIVE));
+
+            // That means the Connection is no longer closeable.
+            assertThat(logicalConnection.isCloseable(), is(false));
+
+            // Ensure JDBC constructs' backlinks work correctly.
+            try (Statement s = logicalConnection.createStatement()) {
+                assertThat(s, notNullValue());
+                assertThat(s.getConnection(), sameInstance(logicalConnection));
+                try (ResultSet rs = s.executeQuery("SHOW TABLES")) {
+                    assertThat(rs.getStatement(), sameInstance(s));
+                }
+            }
+
+            // close() should "close" the logical connection, but not close the physical connection.
+            logicalConnection.close();
+            assertThat(logicalConnection.isClosed(), is(true));
+            assertThat(physicalConnection.isClosed(), is(false));
+
+            // Make sure a close() attempt was recorded.
+            assertThat(logicalConnection.isClosePending(), is(true));
+
+            // But "closing" should have disassociated the XAResource.
+            assertThat(ASSOCIATIONS.get(xid).branchState(), is(IDLE));
+
+            // What happens when we do re-enlisting behavior? First, we'd better appear closed since we called close()
+            // above:
+            assertThrows(SQLException.class, () -> logicalConnection.getHoldability());
+
+            // Let's reenable closeability for this test, which is something users won't do:
+            logicalConnection.setCloseable(true);
+            assertThat(logicalConnection.isCloseable(), is(true));
+            assertThat(logicalConnection.isClosePending(), is(false));
+            assertThat(logicalConnection.isClosed(), is(false));
+
+            // Assert after all of this we're still IDLE, because we don't remove the XAResource from the
+            // TransactionSynchronizationRegistry, so the enlisted() method thinks we're already enlisted, so
+            // getHoldability() skips enlistment, so everything stays as it was.
+            assertThat(ASSOCIATIONS.get(xid).branchState(), is(IDLE));
+
+            // Commit (we didn't actually do any work) AND DISASSOCIATE the transaction, which can only happen with a
+            // call to TransactionManager.commit(), not just Transaction.commit().
+            tm.commit();
+
+            // Transaction is over; the connection should be closeable again.
+            assertThat(logicalConnection.isCloseable(), is(true));
+
+            // Transaction is over; make sure the XAResource removed the association.
+            assertThat(ASSOCIATIONS.size(), is(0));
+
+            // We should be able to actually close it early.  The auto-close should not fail, either.
+            logicalConnection.close();
+            assertThat(logicalConnection.isClosed(), is(true));
+            assertThat(logicalConnection.delegate().isClosed(), is(true));
+        }
+
+        LOGGER.info("Ending testSpike()");
+    }
+
+    @Test
+    final void testTimeout() throws InterruptedException, NotSupportedException, RollbackException, SystemException {
+        LOGGER.info("Starting testTimeout()");
+
+        tm.setTransactionTimeout(1); // 1 second; the minimum settable value (0 means "use the default" (!))
+        tm.begin();
+
+        // For this test, where transaction reaping is set to happen soon, it still won't happen in under 1000
+        // milliseconds, so this assertion is OK unless the testing environment is completely pathological. For
+        // real-world scenarios, you shouldn't assume anything about the initial status.
+        assertThat(tm.getStatus(), is(Status.STATUS_ACTIVE));
+
+        CountDownLatch latch = new CountDownLatch(1);
+        Thread mainThread = Thread.currentThread();
+
+        tsr.registerInterposedSynchronization(new Synchronization() {
+                public void beforeCompletion() {
+
+                }
+                public void afterCompletion(int status) {
+                    try {
+                        assertThat(status, is(Status.STATUS_ROLLEDBACK));
+                        assertThat(Thread.currentThread(), not(mainThread));
+                    } finally {
+                        latch.countDown();
+                    }
+                }
+            });
+
+        // Wait for the transaction to roll back on the reaper thread.  The transaction timeout is 1 second (see above);
+        // this waits for 2 seconds max.  If this fails with an InterruptedException, which should be impossible, check
+        // the logs for Narayana warning that the assertions in the synchronization above failed.
+        latch.await(2000L, TimeUnit.MILLISECONDS);
+
+        // In this case, we never issued a rollback ourselves and we never acquired a Transaction.  Here we show that
+        // you can get a Transaction that is initially in a rolled back state.
+        //
+        // Verify there *was* a transaction but now it is rolled back, so is essentially useless other than as a
+        // tombstone.
+        Transaction t = tm.getTransaction();
+        assertThat(t, notNullValue());
+        assertThat(t.getStatus(), is(Status.STATUS_ROLLEDBACK));
+
+        // Verify that all the other status accessors return the same thing.
+        assertThat(tm.getStatus(), is(Status.STATUS_ROLLEDBACK));
+        assertThat(tsr.getTransactionStatus(), is(Status.STATUS_ROLLEDBACK));
+
+        // Verify that indeed you cannot enlist any XAResource in the transaction when it is in the rolled back state.
+        assertThrows(IllegalStateException.class, () -> t.enlistResource(new NoOpXAResource()));
+
+        // Verify that even though the current transaction is rolled back you can still roll it back again (no-op) and
+        // disassociate it from the current thread.
+        tm.rollback();
+        assertThat(tm.getStatus(), is(Status.STATUS_NO_TRANSACTION));
+        assertThat(tsr.getTransactionStatus(), is(Status.STATUS_NO_TRANSACTION));
+
+        // The Transaction itself remains in status Status.STATUS_ROLLEDBACK. Notably it never enters
+        // Status.STATUS_NO_TRANSACTION.
+        assertThat(t.getStatus(), is(Status.STATUS_ROLLEDBACK));
+
+        assertThat(ASSOCIATIONS.size(), is(0));
+
+        LOGGER.info("Ending testTimeout()");
+    }
+
+    @Test
+    final void testBeginSuspendBeginCommitResumeCommit()
+        throws HeuristicMixedException,
+               HeuristicRollbackException,
+               InvalidTransactionException,
+               NotSupportedException,
+               RollbackException,
+               SQLException,
+               SystemException {
+        LOGGER.info("Starting testBeginSuspendBeginCommitResumeCommit()");
+
+        tm.begin();
+
+        Transaction t = tm.getTransaction();
+        assertThat(t.getStatus(), is(Status.STATUS_ACTIVE));
+
+        Connection physicalConnection = h2ds.getConnection();
+        JtaConnection logicalConnection =
+            new JtaConnection(tm::getTransaction,
+                              tsr,
+                              true,
+                              null,
+                              physicalConnection,
+                              true);
+
+        assertThat(logicalConnection.enlisted(), is(true));
+        assertThat(logicalConnection.delegate(), sameInstance(physicalConnection));
+
+        // Suspend the current transaction.  It will stay in Status.STATUS_ACTIVE state, because suspension has no
+        // effect on the actual state of the *Transaction*, only on the state of its association with the current
+        // thread.
+        Transaction s = tm.suspend();
+        assertThat(s, sameInstance(t));
+        assertThat(s.getStatus(), is(Status.STATUS_ACTIVE));
+
+        // The TransactionManager will report that there is no transaction.
+        assertThat(tm.getStatus(), is(Status.STATUS_NO_TRANSACTION));
+
+        // assertThat(logicalConnection.enlisted(), is(true)); // we're enlisted in a suspended transaction
+        assertThat(logicalConnection.isCloseable(), is(false)); // we're still enlisted in a suspended transaction
+
+        logicalConnection.close(); // doesn't really close, but the caller thinks it did, which is what we want
+
+        // We record that a close was pending.
+        assertThat(logicalConnection.isClosePending(), is(true));
+
+        // A pending close looks to the caller like a real one.
+        assertThat(logicalConnection.isClosed(), is(true));
+
+        // But it's not.
+        assertThat(physicalConnection.isClosed(), is(false));
+
+        tm.begin();
+        t = tm.getTransaction();
+        assertThat(t, not(s));
+        assertThat(t.getStatus(), is(Status.STATUS_ACTIVE));
+        assertThat(s.getStatus(), is(Status.STATUS_ACTIVE));
+        assertThat(tm.getStatus(), is(Status.STATUS_ACTIVE));
+
+        Connection physicalConnection2 = h2ds.getConnection();
+        JtaConnection logicalConnection2 =
+            new JtaConnection(tm::getTransaction,
+                              tsr,
+                              true,
+                              null,
+                              physicalConnection2,
+                              true);
+
+        assertThat(logicalConnection2.enlisted(), is(true));
+        assertThat(logicalConnection2.delegate(), sameInstance(physicalConnection2));
+
+        tm.commit();
+
+        assertThat(t.getStatus(), is(Status.STATUS_COMMITTED));
+        assertThat(tm.getStatus(), is(Status.STATUS_NO_TRANSACTION));
+
+        assertThat(logicalConnection2.enlisted(), is(false)); // we can call this because we haven't closed yet
+
+        assertThat(logicalConnection2.isCloseable(), is(true));
+        logicalConnection2.close();
+        assertThat(logicalConnection2.isClosed(), is(true));
+        assertThat(physicalConnection2.isClosed(), is(true));
+
+        tm.resume(s);
+        
+        t = tm.getTransaction();
+        assertThat(t, sameInstance(s));
+        assertThat(t.getStatus(), is(Status.STATUS_ACTIVE));
+
+        // The first logical connection still looks closed.
+        assertThat(logicalConnection.isClosed(), is(true));
+
+        // But it's not.
+        assertThat(logicalConnection.isClosePending(), is(true));
+        assertThat(physicalConnection.isClosed(), is(false));
+
+        tm.commit();
+
+        assertThat(logicalConnection.isClosePending(), is(true));
+
+        assertThat(logicalConnection.isClosed(), is(true)); // returns true only because close is pending
+        assertThat(logicalConnection.delegate().isClosed(), is(false));
+
+        assertThat(physicalConnection.isClosed(), is(false)); // logicalConnection was never *really* closed
+        assertThat(logicalConnection2.isClosed(), is(true));
+        assertThat(physicalConnection2.isClosed(), is(true));
+
+        LOGGER.info("Ending testBeginSuspendBeginCommitResumeCommit()");
+    }
+
+}
diff --git a/integrations/jta/jdbc/src/test/java/io/helidon/integrations/jta/jdbc/TestTransactionSpecificConnection.java b/integrations/jta/jdbc/src/test/java/io/helidon/integrations/jta/jdbc/TestTransactionSpecificConnection.java
new file mode 100644
index 00000000000..9792ea77f55
--- /dev/null
+++ b/integrations/jta/jdbc/src/test/java/io/helidon/integrations/jta/jdbc/TestTransactionSpecificConnection.java
@@ -0,0 +1,165 @@
+/*
+ * Copyright (c) 2023 Oracle and/or its affiliates.
+ *
+ * Licensed 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 io.helidon.integrations.jta.jdbc;
+
+import java.lang.reflect.Field;
+import java.sql.Connection;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+
+import javax.sql.DataSource;
+
+import com.zaxxer.hikari.HikariConfig;
+import com.zaxxer.hikari.HikariDataSource;
+import com.zaxxer.hikari.pool.ProxyConnection;
+import jakarta.transaction.HeuristicMixedException;
+import jakarta.transaction.HeuristicRollbackException;
+import jakarta.transaction.NotSupportedException;
+import jakarta.transaction.RollbackException;
+import jakarta.transaction.Status;
+import jakarta.transaction.SystemException;
+import jakarta.transaction.Transaction;
+import jakarta.transaction.TransactionManager;
+import org.h2.jdbcx.JdbcDataSource;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.not;
+import static org.hamcrest.CoreMatchers.sameInstance;
+import static org.hamcrest.MatcherAssert.assertThat;
+
+@Deprecated(forRemoval = true, since = "3.1.1")
+final class TestTransactionSpecificConnection {
+
+    private JdbcDataSource h2ds;
+
+    private TransactionManager tm;
+
+    private TestTransactionSpecificConnection() {
+        super();
+    }
+
+    @BeforeEach
+    void initializeH2DataSource() throws SQLException, SystemException {
+        JdbcDataSource ds = new JdbcDataSource();
+        ds.setURL("jdbc:h2:mem:test");
+        ds.setUser("sa");
+        ds.setPassword("sa");
+        this.h2ds = ds;
+        this.tm = com.arjuna.ats.jta.TransactionManager.transactionManager();
+        this.tm.setTransactionTimeout(20 * 60); // 20 minutes for debugging
+    }
+
+    @Test
+    void testConnectionPoolSemantics() throws IllegalAccessException, NoSuchFieldException, SQLException {
+        Field delegate = ProxyConnection.class.getDeclaredField("delegate");
+        assertThat(delegate.trySetAccessible(), is(true));
+        HikariConfig hc = new HikariConfig();
+        hc.setDataSource(this.h2ds);
+        try (HikariDataSource ds = new HikariDataSource(hc)) {
+            Connection c1 = ds.getConnection();
+            Connection d1 = (Connection) delegate.get(c1);
+            c1.close();
+            assertThat(c1.isClosed(), is(true));
+            assertThat(d1.isClosed(), is(false));
+
+            Connection c2 = ds.getConnection();
+            Connection d2 = (Connection) delegate.get(c2);
+            c2.close();
+            assertThat(c2.isClosed(), is(true));
+            assertThat(c1, not(sameInstance(c2)));
+            assertThat(d1, sameInstance(d2));
+        }
+    }
+
+    @Test
+    void testCloseableAndClosedBehavior() throws SQLException {
+        @SuppressWarnings("removal")
+        io.helidon.integrations.jta.jdbc.JtaDataSource.TransactionSpecificConnection c =
+            new io.helidon.integrations.jta.jdbc.JtaDataSource.TransactionSpecificConnection(this.h2ds.getConnection());
+
+        assertThat(c.isCloseable(), is(false));
+        assertThat(c.isClosed(), is(false));
+        assertThat(c.getAutoCommit(), is(false));
+
+        c.close(); // no-op
+        assertThat(c.isCloseable(), is(false));
+        assertThat(c.isClosed(), is(false));
+        assertThat(c.isCloseCalled(), is(true));
+
+        c.setCloseable(true);
+        assertThat(c.isCloseable(), is(true));
+        assertThat(c.isCloseCalled(), is(true)); // still
+        assertThat(c.isClosed(), is(false));
+
+        c.close(); // the real thing
+        assertThat(c.isClosed(), is(true));
+        assertThat(c.isCloseCalled(), is(true));
+        assertThat(c.isCloseable(), is(false));
+    }
+
+    @Test
+    void testTransactionManagerSemantics()
+        throws HeuristicMixedException,
+               HeuristicRollbackException,
+               NotSupportedException,
+               RollbackException,
+               SQLException,
+               SystemException {
+        TransactionManager tm = com.arjuna.ats.jta.TransactionManager.transactionManager();
+        tm.setTransactionTimeout(20 * 60); // 20 minutes for debugging
+        tm.begin();
+        Transaction t = tm.getTransaction();
+
+        // The TransactionSpecificConnection class does not interact
+        // with the TransactionManager in any way by itself; prove
+        // that this is the case.
+        @SuppressWarnings("removal")
+        io.helidon.integrations.jta.jdbc.JtaDataSource.TransactionSpecificConnection c
+            = new io.helidon.integrations.jta.jdbc.JtaDataSource.TransactionSpecificConnection(this.h2ds.getConnection());
+
+        // (The jakarta.transaction.Transaction/TransactionManager
+        // state machine is weirder than weird.)
+        t.commit(); // unnecessary
+        tm.commit();
+
+        // Committing the Transaction results in STATUS_COMMITTED.
+        // Makes sense.
+        assertThat(t.getStatus(), is(Status.STATUS_COMMITTED));
+
+        // Committing the TransactionManager results in
+        // STATUS_NO_TRANSACTION.
+        // (https://jakarta.ee/specifications/transactions/2.0/jakarta-transactions-spec-2.0.html#completing-a-transaction).
+        // Also makes sense.  What may not immediately make sense is
+        // why they're not the same state. See
+        // https://groups.google.com/g/narayana-users/c/eYVUmhE9QZg.
+        assertThat(tm.getStatus(), is(Status.STATUS_NO_TRANSACTION));
+
+        assertThat(c.isCloseable(), is(false));
+
+        c.close();
+        assertThat(c.isClosed(), is(false));
+
+        c.setCloseable(true);
+        assertThat(c.isCloseable(), is(true));
+
+        c.close();
+        assertThat(c.isClosed(), is(true));
+    }
+
+}