ISPN-10924 JPA cache store (with async enabled) will log Exceptions i…

…n DEBUG for deleteBatch and is not able to remove entries from the store

persistence/jpa/src/main/java/org/infinispan/persistence/jpa/JpaStore.java

@@ -343,11 +343,12 @@ public void deleteBatch(Iterable<Object> keys) {
             CriteriaDelete query = cb.createCriteriaDelete(configuration.entityClass());
             Root root = query.from(configuration.entityClass());
             SingularAttribute id = getEntityId(em, configuration.entityClass());
-            query.where(root.get(id).in(keys));
+            List<Object> keyCollection = StreamSupport.stream(keys.spliterator(), false).collect(Collectors.toList());
+            query.where(root.get(id).in(keyCollection));
             em.createQuery(query).executeUpdate();
 
             if (configuration.storeMetadata()) {
-               List<MetadataEntityKey> metaKeys = StreamSupport.stream(keys.spliterator(), false).map(this::getMetadataKey).collect(Collectors.toList());
+               List<MetadataEntityKey> metaKeys = keyCollection.stream().map(this::getMetadataKey).collect(Collectors.toList());
                CriteriaDelete<MetadataEntity> metaQuery = cb.createCriteriaDelete(MetadataEntity.class);
                Root<MetadataEntity> metaRoot = metaQuery.from(MetadataEntity.class);
                id = getEntityId(em, MetadataEntity.class);

persistence/jpa/src/test/java/org/infinispan/persistence/jpa/JpaStoreAsyncFunctionalTest.java

@@ -0,0 +1,44 @@
+package org.infinispan.persistence.jpa;
+
+import org.infinispan.Cache;
+import org.infinispan.configuration.cache.ConfigurationBuilder;
+import org.infinispan.configuration.global.GlobalConfigurationBuilder;
+import org.infinispan.manager.EmbeddedCacheManager;
+import org.infinispan.persistence.jpa.configuration.JpaStoreConfigurationBuilder;
+import org.infinispan.persistence.jpa.entity.KeyValueEntity;
+import org.infinispan.test.SingleCacheManagerTest;
+import org.infinispan.test.TestingUtil;
+import org.infinispan.test.fwk.TestCacheManagerFactory;
+import org.testng.annotations.Test;
+
+@Test(groups = "functional", testName = "persistence.JpaStoreAsyncFunctionalTest")
+public class JpaStoreAsyncFunctionalTest extends SingleCacheManagerTest {
+   @Override
+   protected EmbeddedCacheManager createCacheManager() {
+      GlobalConfigurationBuilder global = new GlobalConfigurationBuilder();
+      global.globalState().persistentLocation(TestingUtil.tmpDirectory(this.getClass()));
+      global.serialization().addContextInitializer(JpaSCI.INSTANCE);
+      return TestCacheManagerFactory.newDefaultCacheManager(false, global, new ConfigurationBuilder(), true);
+   }
+
+   private JpaStoreConfigurationBuilder createJpaConfig() {
+      return new ConfigurationBuilder().persistence()
+         .addStore(JpaStoreConfigurationBuilder.class)
+            .persistenceUnitName("org.infinispan.persistence.jpa")
+            .entityClass(KeyValueEntity.class)
+            .segmented(false);
+   }
+
+   public void testAsyncWriteAndDelete() {
+      cacheManager.defineConfiguration("ASYNC_STORE", createJpaConfig().async().enable().build());
+      cacheManager.defineConfiguration("SYNC_STORE", createJpaConfig().build());
+      Cache<String, KeyValueEntity> asyncStore = cacheManager.getCache("ASYNC_STORE");
+      JpaStore<String, KeyValueEntity> syncStore = TestingUtil.getFirstWriter(cacheManager.getCache("SYNC_STORE"));
+
+      String key = "1";
+      asyncStore.put(key, new KeyValueEntity(key, "Example"));
+      eventually(() -> syncStore.contains(key));
+      asyncStore.remove(key);
+      eventually(() -> !syncStore.contains(key));
+   }
+}