ISPN-9569 Indexing is not working for clustered caches if it is trans…

…actional

* TransactionalIndexingTest highlights the issue
* remove ProtobufValueWrapperInterceptor; its role is now taken by ProtobufValueWrapperSearchWorkCreator
* move ProtobufValueWrapperAnalyzerDiscriminator and ProtobufValueWrapperIndexingInterceptor
  to org.infinispan.query.remote.impl.indexing package

client/hotrod-client/src/test/java/org/infinispan/client/hotrod/query/MultiHotRodServerQueryTest.java

@@ -44,9 +44,13 @@ public class MultiHotRodServerQueryTest extends MultiHotRodServersTest {
    protected RemoteCache<Integer, User> remoteCache0;
    protected RemoteCache<Integer, User> remoteCache1;
 
+   protected boolean useTransactions() {
+      return false;
+   }
+
    @Override
    protected void createCacheManagers() throws Throwable {
-      ConfigurationBuilder builder = hotRodCacheConfiguration(getDefaultClusteredCacheConfig(CacheMode.REPL_SYNC, false));
+      ConfigurationBuilder builder = hotRodCacheConfiguration(getDefaultClusteredCacheConfig(CacheMode.REPL_SYNC, useTransactions()));
       builder.indexing().index(Index.ALL)
             .addProperty("default.directory_provider", "local-heap")
             .addProperty("lucene_version", "LUCENE_CURRENT");
@@ -113,7 +117,7 @@ protected void populateCache() throws Exception {
       client(0).getCache().put("dummy", "a primitive value cannot be queried");
    }
 
-   public void testAttributeQuery() throws Exception {
+   public void testAttributeQuery() {
       // get user back from remote cache and check its attributes
       User fromCache = remoteCache0.get(1);
       assertNotNull(fromCache);
@@ -131,7 +135,7 @@ public void testAttributeQuery() throws Exception {
       assertUser1(list.get(0));
    }
 
-   public void testGroupByQuery() throws Exception {
+   public void testGroupByQuery() {
       // get user back from remote cache and check its attributes
       User fromCache = remoteCache0.get(1);
       assertNotNull(fromCache);
@@ -154,7 +158,7 @@ public void testGroupByQuery() throws Exception {
       assertEquals("Tom", list.get(1)[0]);
    }
 
-   public void testEmbeddedAttributeQuery() throws Exception {
+   public void testEmbeddedAttributeQuery() {
       // get user back from remote cache via query and check its attributes
       QueryFactory qf = Search.getQueryFactory(remoteCache1);
       Query query = qf.from(UserPB.class)
@@ -168,7 +172,7 @@ public void testEmbeddedAttributeQuery() throws Exception {
    }
 
    @Test(expectedExceptions = HotRodClientException.class, expectedExceptionsMessageRegExp = ".*ISPN028503: Property addresses can not be selected from type sample_bank_account.User since it is an embedded entity.")
-   public void testInvalidEmbeddedAttributeQuery() throws Exception {
+   public void testInvalidEmbeddedAttributeQuery() {
       QueryFactory qf = Search.getQueryFactory(remoteCache1);
 
       Query q = qf.from(UserPB.class)
@@ -177,7 +181,7 @@ public void testInvalidEmbeddedAttributeQuery() throws Exception {
       q.list();  // exception expected
    }
 
-   public void testProjections() throws Exception {
+   public void testProjections() {
       // get user back from remote cache and check its attributes
       User fromCache = remoteCache0.get(1);
       assertUser1(fromCache);

client/hotrod-client/src/test/java/org/infinispan/client/hotrod/query/TransactionalIndexingTest.java

@@ -0,0 +1,15 @@
+package org.infinispan.client.hotrod.query;
+
+import org.testng.annotations.Test;
+
+/**
+ * @since 9.4
+ */
+@Test(testName = "client.hotrod.query.TransactionalIndexingTest", groups = "functional")
+public class TransactionalIndexingTest extends MultiHotRodServerQueryTest {
+
+   @Override
+   protected boolean useTransactions() {
+      return true;
+   }
+}

query/src/main/java/org/infinispan/query/backend/ExtendedSearchWorkCreator.java

@@ -1,10 +1,14 @@
 package org.infinispan.query.backend;
 
+//todo [anistor] this class will be removed in Infinispan 10
+
 /**
  * Add extra methods.
  *
  * @author Ales Justin
+ * @deprecated without replacement
  */
+@Deprecated
 public interface ExtendedSearchWorkCreator extends SearchWorkCreator {
    boolean shouldRemove(SearchWorkCreatorContext context);
 }

query/src/main/java/org/infinispan/query/backend/QueryInterceptor.java

@@ -329,7 +329,7 @@ public void purgeIndex(Class<?> entityType) {
    }
 
    /**
-    * Remove entries from all indexes by key
+    * Remove entries from all indexes by key.
     */
    void removeFromIndexes(TransactionContext transactionContext, Object key) {
       Stream<IndexedTypeIdentifier> typeIdentifiers = getKnownClasses().stream()

query/src/main/java/org/infinispan/query/backend/SearchWorkCreatorContext.java

@@ -1,10 +1,14 @@
 package org.infinispan.query.backend;
 
+//todo [anistor] this class will be removed in Infinispan 10
+
 /**
  * Use context pattern, so it can be easily extended / changed.
  *
  * @author Ales Justin
+ * @deprecated without replacement
  */
+@Deprecated
 public class SearchWorkCreatorContext {
    private Object previousValue;
    private Object currentValue;

remote-query/remote-query-server/src/main/java/org/infinispan/query/remote/impl/LifecycleManager.java

@@ -18,26 +18,20 @@
 import org.infinispan.commons.marshall.AdvancedExternalizer;
 import org.infinispan.commons.util.ServiceFinder;
 import org.infinispan.configuration.cache.Configuration;
-import org.infinispan.configuration.cache.ConfigurationBuilder;
 import org.infinispan.configuration.cache.ContentTypeConfiguration;
-import org.infinispan.configuration.cache.CustomInterceptorsConfigurationBuilder;
-import org.infinispan.configuration.cache.InterceptorConfiguration;
-import org.infinispan.configuration.cache.InterceptorConfigurationBuilder;
 import org.infinispan.configuration.global.GlobalConfiguration;
 import org.infinispan.configuration.global.GlobalJmxStatisticsConfiguration;
 import org.infinispan.factories.ComponentRegistry;
 import org.infinispan.factories.GlobalComponentRegistry;
 import org.infinispan.factories.components.ComponentMetadataRepo;
 import org.infinispan.factories.components.ManageableComponentMetadata;
 import org.infinispan.factories.impl.BasicComponentRegistry;
-import org.infinispan.interceptors.AsyncInterceptorChain;
-import org.infinispan.interceptors.impl.BatchingInterceptor;
-import org.infinispan.interceptors.impl.InvocationContextInterceptor;
 import org.infinispan.jmx.ResourceDMBean;
 import org.infinispan.lifecycle.ModuleLifecycle;
 import org.infinispan.manager.EmbeddedCacheManager;
 import org.infinispan.marshall.core.EncoderRegistry;
 import org.infinispan.protostream.SerializationContext;
+import org.infinispan.query.backend.QueryInterceptor;
 import org.infinispan.query.remote.ProtobufMetadataManager;
 import org.infinispan.query.remote.client.Externalizers.QueryRequestExternalizer;
 import org.infinispan.query.remote.client.ProtostreamSerializationContextInitializer;
@@ -53,7 +47,7 @@
 import org.infinispan.query.remote.impl.filter.IckleProtobufCacheEventFilterConverter;
 import org.infinispan.query.remote.impl.filter.IckleProtobufFilterAndConverter;
 import org.infinispan.query.remote.impl.indexing.ProtobufValueWrapper;
-import org.infinispan.query.remote.impl.indexing.ProtobufValueWrapperInterceptor;
+import org.infinispan.query.remote.impl.indexing.ProtobufValueWrapperSearchWorkCreator;
 import org.infinispan.query.remote.impl.logging.Log;
 import org.infinispan.registry.InternalCacheRegistry;
 import org.kohsuke.MetaInfServices;
@@ -174,40 +168,6 @@ public void cacheStarting(ComponentRegistry cr, Configuration cfg, String cacheN
          ProtobufMetadataManagerImpl protobufMetadataManager =
             (ProtobufMetadataManagerImpl) gcr.getComponent(ProtobufMetadataManager.class).running();
          protobufMetadataManager.addCacheDependency(cacheName);
-
-         if (cfg.indexing().index().isEnabled()) {
-            log.infof("Registering ProtobufValueWrapperInterceptor for cache %s", cacheName);
-            EmbeddedCacheManager cacheManager = gcr.getComponent(EmbeddedCacheManager.class).running();
-            createProtobufValueWrapperInterceptor(cr, cfg, cacheManager);
-         }
-      }
-   }
-
-   private void createProtobufValueWrapperInterceptor(ComponentRegistry cr, Configuration cfg, EmbeddedCacheManager cacheManager) {
-      ProtobufValueWrapperInterceptor wrapperInterceptor = cr.getComponent(ProtobufValueWrapperInterceptor.class);
-      if (wrapperInterceptor == null) {
-         SerializationContext serCtx = ProtobufMetadataManagerImpl.getSerializationContext(cacheManager);
-         wrapperInterceptor = new ProtobufValueWrapperInterceptor(serCtx);
-
-         // Interceptor registration not needed, core configuration handling
-         // already does it for all custom interceptors - UNLESS the InterceptorChain already exists in the component registry!
-         AsyncInterceptorChain ic = cr.getComponent(AsyncInterceptorChain.class);
-
-         ConfigurationBuilder builder = new ConfigurationBuilder().read(cfg);
-         InterceptorConfigurationBuilder interceptorBuilder = builder.customInterceptors().addInterceptor();
-         interceptorBuilder.interceptor(wrapperInterceptor);
-
-         if (cfg.invocationBatching().enabled()) {
-            if (ic != null) ic.addInterceptorAfter(wrapperInterceptor, BatchingInterceptor.class);
-            interceptorBuilder.after(BatchingInterceptor.class);
-         } else {
-            if (ic != null) ic.addInterceptorAfter(wrapperInterceptor, InvocationContextInterceptor.class);
-            interceptorBuilder.after(InvocationContextInterceptor.class);
-         }
-         if (ic != null) {
-            cr.registerComponent(wrapperInterceptor, ProtobufValueWrapperInterceptor.class);
-         }
-         cfg.customInterceptors().interceptors(builder.build().customInterceptors().interceptors());
       }
    }
 
@@ -241,45 +201,19 @@ public void cacheStarted(ComponentRegistry cr, String cacheName) {
       InternalCacheRegistry icr = cr.getGlobalComponentRegistry().getComponent(InternalCacheRegistry.class);
       if (!icr.isInternalCache(cacheName)) {
          Configuration cfg = cr.getComponent(Configuration.class);
-         if (cfg.indexing().index().isEnabled()) {
-            if (!verifyChainContainsProtobufValueWrapperInterceptor(cr)) {
-               throw new IllegalStateException("It was expected to find the ProtobufValueWrapperInterceptor registered in the InterceptorChain but it wasn't found");
-            }
-         } else if (verifyChainContainsProtobufValueWrapperInterceptor(cr)) {
-            throw new IllegalStateException("It was NOT expected to find the ProtobufValueWrapperInterceptor registered in the InterceptorChain as indexing was disabled, but it was found");
-         }
-
          ProtobufMetadataManagerImpl protobufMetadataManager = (ProtobufMetadataManagerImpl) cr.getGlobalComponentRegistry().getComponent(ProtobufMetadataManager.class);
          SerializationContext serCtx = protobufMetadataManager.getSerializationContext();
 
          RemoteQueryManager remoteQueryManager = buildQueryManager(cfg, serCtx, cr);
 
          cr.registerComponent(remoteQueryManager, RemoteQueryManager.class);
-      }
-   }
-
-   private boolean verifyChainContainsProtobufValueWrapperInterceptor(ComponentRegistry cr) {
-      AsyncInterceptorChain interceptorChain = cr.getComponent(AsyncInterceptorChain.class);
-      return interceptorChain != null && interceptorChain.containsInterceptorType(ProtobufValueWrapperInterceptor.class, true);
-   }
 
-   @Override
-   public void cacheStopped(ComponentRegistry cr, String cacheName) {
-      Configuration cfg = cr.getComponent(Configuration.class);
-      removeProtobufValueWrapperInterceptor(cfg);
-   }
-
-   private void removeProtobufValueWrapperInterceptor(Configuration cfg) {
-      ConfigurationBuilder builder = new ConfigurationBuilder();
-      CustomInterceptorsConfigurationBuilder customInterceptorsBuilder = builder.customInterceptors();
-
-      for (InterceptorConfiguration interceptorConfig : cfg.customInterceptors().interceptors()) {
-         if (!(interceptorConfig.asyncInterceptor() instanceof ProtobufValueWrapperInterceptor)) {
-            customInterceptorsBuilder.addInterceptor().read(interceptorConfig);
+         if (cfg.indexing().index().isEnabled()) {
+            log.debugf("Wrapping the SearchWorkCreator for indexed cache %s", cacheName);
+            QueryInterceptor queryInterceptor = cr.getComponent(QueryInterceptor.class);
+            queryInterceptor.setSearchWorkCreator(new ProtobufValueWrapperSearchWorkCreator(queryInterceptor.getSearchWorkCreator(), serCtx).get());
          }
       }
-
-      cfg.customInterceptors().interceptors(builder.build().customInterceptors().interceptors());
    }
 
    @Override

remote-query/remote-query-server/src/main/java/org/infinispan/query/remote/impl/ProgrammaticSearchMappingProviderImpl.java

@@ -7,7 +7,9 @@
 import org.infinispan.Cache;
 import org.infinispan.commons.logging.LogFactory;
 import org.infinispan.query.remote.impl.indexing.ProtobufValueWrapper;
+import org.infinispan.query.remote.impl.indexing.ProtobufValueWrapperAnalyzerDiscriminator;
 import org.infinispan.query.remote.impl.indexing.ProtobufValueWrapperFieldBridge;
+import org.infinispan.query.remote.impl.indexing.ProtobufValueWrapperIndexingInterceptor;
 import org.infinispan.query.remote.impl.logging.Log;
 import org.infinispan.query.spi.ProgrammaticSearchMappingProvider;
 import org.kohsuke.MetaInfServices;

remote-query/remote-query-server/src/main/java/org/infinispan/query/remote/impl/indexing/ProtobufValueWrapper.java

@@ -26,14 +26,24 @@ public final class ProtobufValueWrapper implements WrappedBytes {
 
    public static final IndexedTypeIdentifier INDEXING_TYPE = PojoIndexedTypeIdentifier.convertFromLegacy(ProtobufValueWrapper.class);
 
+   /**
+    * Max number of bytes to include in {@link #toString()}.
+    */
    private static final int MAX_BYTES_IN_TOSTRING = 40;
 
-   // The protobuf encoded payload
+   /**
+    * The protobuf encoded payload.
+    */
    private final byte[] binary;
 
+   /**
+    * The lazily computed hashCode. Transient field!
+    */
    private transient int hashCode = 0;
 
-   // The Descriptor of the message (if it's a Message and not a primitive value). Transient field!
+   /**
+    * The Descriptor of the message (if it's a Message and not a primitive value, or null otherwise). Transient field!
+    */
    private transient Descriptor messageDescriptor;
 
    public ProtobufValueWrapper(byte[] binary) {
@@ -43,21 +53,30 @@ public ProtobufValueWrapper(byte[] binary) {
       this.binary = binary;
    }
 
+   /**
+    * Gets the internal byte array. Callers should not modify the contents of the array.
+    *
+    * @return the wrapped byte array
+    */
    public byte[] getBinary() {
       return binary;
    }
 
    /**
     * Returns the Protobuf descriptor of the message type of the payload.
+    *
+    * @see #setMessageDescriptor
     */
    public Descriptor getMessageDescriptor() {
       return messageDescriptor;
    }
 
    /**
     * Sets the Protobuf descriptor of the message type of the payload.
+    *
+    * @see ProtobufValueWrapperSearchWorkCreator#discoverMessageType
     */
-   public void setMessageDescriptor(Descriptor messageDescriptor) {
+   void setMessageDescriptor(Descriptor messageDescriptor) {
       this.messageDescriptor = messageDescriptor;
    }
 
@@ -117,6 +136,16 @@ public byte getByte(int offset) {
       return binary[offset];
    }
 
+   @Override
+   public boolean equalsWrappedBytes(WrappedBytes other) {
+      if (other == null) return false;
+      if (other.getLength() != binary.length) return false;
+      for (int i = 0; i < binary.length; ++i) {
+         if (binary[i] != other.getByte(i)) return false;
+      }
+      return true;
+   }
+
    public static final class Externalizer extends AbstractExternalizer<ProtobufValueWrapper> {
 
       @Override

remote-query/remote-query-server/src/main/java/org/infinispan/query/remote/impl/ProtobufValueWrapperAnalyzerDiscriminator.java → remote-query/remote-query-server/src/main/java/org/infinispan/query/remote/impl/indexing/ProtobufValueWrapperAnalyzerDiscriminator.java

@@ -1,11 +1,8 @@
-package org.infinispan.query.remote.impl;
+package org.infinispan.query.remote.impl.indexing;
 
 import org.hibernate.search.analyzer.Discriminator;
 import org.infinispan.protostream.descriptors.Descriptor;
 import org.infinispan.query.logging.Log;
-import org.infinispan.query.remote.impl.indexing.FieldMapping;
-import org.infinispan.query.remote.impl.indexing.IndexingMetadata;
-import org.infinispan.query.remote.impl.indexing.ProtobufValueWrapper;
 import org.infinispan.util.logging.LogFactory;
 
 /**
@@ -25,6 +22,10 @@ public String getAnalyzerDefinitionName(Object value, Object entity, String fiel
          Descriptor messageDescriptor = wrapper.getMessageDescriptor();
          if (messageDescriptor != null) {
             return getAnalyzerForField(messageDescriptor, field);
+         } else {
+            // this is either a scalar value (not indexed, why are we asked for analyzer -> bug)
+            // or this entry was not run through ProtobufValueWrapperSearchWorkCreator (bug again)
+            throw new IllegalStateException("Message descriptor not initialized for " + wrapper);
          }
       }
       return null;

remote-query/remote-query-server/src/main/java/org/infinispan/query/remote/impl/ProtobufValueWrapperIndexingInterceptor.java → remote-query/remote-query-server/src/main/java/org/infinispan/query/remote/impl/indexing/ProtobufValueWrapperIndexingInterceptor.java

@@ -1,10 +1,8 @@
-package org.infinispan.query.remote.impl;
+package org.infinispan.query.remote.impl.indexing;
 
 import org.hibernate.search.indexes.interceptor.EntityIndexingInterceptor;
 import org.hibernate.search.indexes.interceptor.IndexingOverride;
 import org.infinispan.protostream.descriptors.Descriptor;
-import org.infinispan.query.remote.impl.indexing.IndexingMetadata;
-import org.infinispan.query.remote.impl.indexing.ProtobufValueWrapper;
 
 /**
  * Hibernate Search interceptor for conditional indexing of protobuf message types based on the value of the