Use RESOURCE_LOCAL transactions for JPA map storage

Closes #15248

quarkus/deployment/src/main/java/org/keycloak/quarkus/deployment/KeycloakProcessor.java

@@ -20,6 +20,7 @@ package org.keycloak.quarkus.deployment;
 import static org.keycloak.quarkus.runtime.KeycloakRecorder.DEFAULT_HEALTH_ENDPOINT;
 import static org.keycloak.quarkus.runtime.KeycloakRecorder.DEFAULT_METRICS_ENDPOINT;
 import static org.keycloak.quarkus.runtime.Providers.getProviderManager;
+import static org.keycloak.quarkus.runtime.configuration.Configuration.getConfig;
 import static org.keycloak.quarkus.runtime.configuration.Configuration.getPropertyNames;
 import static org.keycloak.quarkus.runtime.configuration.MicroProfileConfigProvider.NS_QUARKUS;
 import static org.keycloak.quarkus.runtime.configuration.QuarkusPropertiesConfigSource.QUARKUS_PROPERTY_ENABLED;
@@ -49,6 +50,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.Objects;
 import java.util.Optional;
 import java.util.Properties;
 import java.util.function.Consumer;
@@ -81,6 +83,7 @@ import io.smallrye.config.ConfigValue;
 import org.hibernate.cfg.AvailableSettings;
 import org.hibernate.jpa.boot.internal.ParsedPersistenceXmlDescriptor;
 import org.hibernate.jpa.boot.internal.PersistenceXmlParser;
+import org.hibernate.resource.jdbc.spi.PhysicalConnectionHandlingMode;
 import org.jboss.jandex.AnnotationInstance;
 import org.jboss.jandex.AnnotationTarget;
 import org.jboss.jandex.ClassInfo;
@@ -93,6 +96,7 @@ import org.keycloak.Config;
 import org.keycloak.common.crypto.FipsMode;
 import org.keycloak.config.SecurityOptions;
 import org.keycloak.config.StorageOptions;
+import org.keycloak.config.TransactionOptions;
 import org.keycloak.connections.jpa.JpaConnectionProvider;
 import org.keycloak.connections.jpa.JpaConnectionSpi;
 import org.keycloak.models.map.storage.jpa.JpaMapStorageProviderFactory;
@@ -273,7 +277,25 @@ class KeycloakProcessor {
         Properties unitProperties = descriptor.getProperties();
 
         unitProperties.setProperty(AvailableSettings.DIALECT, config.defaultPersistenceUnit.dialect.dialect.orElse(null));
-        unitProperties.setProperty(AvailableSettings.JPA_TRANSACTION_TYPE, PersistenceUnitTransactionType.JTA.name());
+        if (Objects.equals(getConfig().getConfigValue("kc.transaction-jta-enabled").getValue(), "disabled")) {
+            unitProperties.setProperty(AvailableSettings.JPA_TRANSACTION_TYPE, PersistenceUnitTransactionType.RESOURCE_LOCAL.name());
+
+            // Only changing this for the new map storage to keep the legacy JPA store untouched as it wasn't tested for the legacy store.
+            // follow-up on this in https://github.com/keycloak/keycloak/issues/13222 to re-visit the auto-commit handling
+            String storage = Configuration.getRawValue(
+                    MicroProfileConfigProvider.NS_KEYCLOAK_PREFIX.concat(StorageOptions.STORAGE.getKey()));
+            if (storage != null) {
+                // Needed to change the connection handling to avoid Hibernate returning the connection too early,
+                // which then interfered with the auto-commit reset that's done at the end of the transaction and PgConnection throwing a "Cannot commit when autoCommit is enabled."
+                // The current default is DELAYED_ACQUISITION_AND_RELEASE_BEFORE_TRANSACTION_COMPLETION which would only make sense for JTA IMHO.
+                // https://github.com/quarkusio/quarkus/blob/8d89101ffa65465b33d06360047095046bb726e4/extensions/hibernate-orm/runtime/src/main/java/io/quarkus/hibernate/orm/runtime/boot/FastBootMetadataBuilder.java#L287-L288
+                unitProperties.setProperty(AvailableSettings.CONNECTION_HANDLING, PhysicalConnectionHandlingMode.DELAYED_ACQUISITION_AND_RELEASE_AFTER_TRANSACTION.name());
+            }
+        } else {
+            // will happen for both "enabled" and "xa"
+            unitProperties.setProperty(AvailableSettings.JPA_TRANSACTION_TYPE, PersistenceUnitTransactionType.JTA.name());
+        }
+
         unitProperties.setProperty(AvailableSettings.QUERY_STARTUP_CHECKING, Boolean.FALSE.toString());
 
         String dbKind = jdbcDataSources.get(0).getDbKind();

quarkus/runtime/src/main/java/org/keycloak/quarkus/runtime/storage/database/jpa/QuarkusJpaMapStorageProviderFactory.java

@@ -17,10 +17,6 @@
 
 package org.keycloak.quarkus.runtime.storage.database.jpa;
 
-import static org.keycloak.config.StorageOptions.STORAGE;
-import static org.keycloak.quarkus.runtime.configuration.Configuration.getOptionalValue;
-import static org.keycloak.quarkus.runtime.configuration.MicroProfileConfigProvider.NS_KEYCLOAK_PREFIX;
-
 import java.lang.annotation.Annotation;
 import java.sql.Connection;
 import java.sql.SQLException;
@@ -31,7 +27,6 @@ import javax.persistence.EntityManagerFactory;
 import org.hibernate.internal.SessionFactoryImpl;
 import org.hibernate.internal.SessionImpl;
 import org.keycloak.config.StorageOptions;
-import org.keycloak.models.ModelException;
 import org.keycloak.models.map.storage.jpa.JpaMapStorageProviderFactory;
 
 import io.quarkus.arc.Arc;
@@ -58,8 +53,7 @@ public class QuarkusJpaMapStorageProviderFactory extends JpaMapStorageProviderFa
     @Override
     protected EntityManager getEntityManager() {
         EntityManager em = super.getEntityManager();
-        try {
+        em.unwrap(SessionImpl.class).doWork(connection -> {
-            Connection connection = em.unwrap(SessionImpl.class).connection();
             // In the Undertow setup, Hibernate sets the connection to non-autocommit, and in the Quarkus setup the XA transaction manager does this.
             // For the Quarkus setup without a XA transaction manager, we didn't find a way to have this setup automatically.
             // There is also no known option to configure this in the Agroal DB connection pool in a Quarkus setup:
@@ -70,9 +64,7 @@ public class QuarkusJpaMapStorageProviderFactory extends JpaMapStorageProviderFa
             if (connection.getAutoCommit()) {
                 connection.setAutoCommit(false);
             }
-        } catch (SQLException e) {
+        });
-            throw new ModelException("unable to set non-auto-commit to false");
-        }
         return em;
     }