HHH-10344 Make transactional invalidation backwards compatible

(cherry picked from commit 05aaeb1963)

hibernate-infinispan/src/main/java/org/hibernate/cache/infinispan/access/PutFromLoadValidator.java

@@ -178,6 +178,11 @@ public class PutFromLoadValidator {
 			}
 			boolean transactional = cache.getCacheConfiguration().transaction().transactionMode().isTransactional();
 			if (transactional) {
+				cache.removeInterceptor(invalidationPosition);
+				TxInvalidationInterceptor txInvalidationInterceptor = new TxInvalidationInterceptor();
+				cache.getComponentRegistry().registerComponent(txInvalidationInterceptor, TxInvalidationInterceptor.class);
+				cache.addInterceptor(txInvalidationInterceptor, invalidationPosition);
+
 				// Note that invalidation does *NOT* acquire locks; therefore, we have to start invalidating before
 				// wrapping the entry, since if putFromLoad was invoked between wrap and beginInvalidatingKey, the invalidation
 				// would not commit the entry removal (as during wrap the entry was not in cache)
@@ -223,6 +228,13 @@ public class PutFromLoadValidator {
 				cache.removeInterceptor(NonTxInvalidationInterceptor.class);
 				break;
 			}
+			else if (i instanceof TxInvalidationInterceptor) {
+				InvalidationInterceptor invalidationInterceptor = new InvalidationInterceptor();
+				cache.getComponentRegistry().registerComponent(invalidationInterceptor, InvalidationInterceptor.class);
+				cache.addInterceptorBefore(invalidationInterceptor, TxInvalidationInterceptor.class);
+				cache.removeInterceptor(TxInvalidationInterceptor.class);
+				break;
+			}
 		}
 		CacheCommandInitializer cci = cache.getComponentRegistry().getComponent(CacheCommandInitializer.class);
 		cci.removePutFromLoadValidator(cache.getName());

hibernate-infinispan/src/main/java/org/hibernate/cache/infinispan/access/TxInvalidationInterceptor.java

@@ -0,0 +1,300 @@
+/*
+ * Hibernate, Relational Persistence for Idiomatic Java
+ *
+ * License: GNU Lesser General Public License (LGPL), version 2.1 or later.
+ * See the lgpl.txt file in the root directory or <http://www.gnu.org/licenses/lgpl-2.1.html>.
+ */
+package org.hibernate.cache.infinispan.access;
+
+import org.infinispan.commands.AbstractVisitor;
+import org.infinispan.commands.CommandsFactory;
+import org.infinispan.commands.FlagAffectedCommand;
+import org.infinispan.commands.ReplicableCommand;
+import org.infinispan.commands.control.LockControlCommand;
+import org.infinispan.commands.tx.PrepareCommand;
+import org.infinispan.commands.write.ClearCommand;
+import org.infinispan.commands.write.InvalidateCommand;
+import org.infinispan.commands.write.PutKeyValueCommand;
+import org.infinispan.commands.write.PutMapCommand;
+import org.infinispan.commands.write.RemoveCommand;
+import org.infinispan.commands.write.ReplaceCommand;
+import org.infinispan.commands.write.WriteCommand;
+import org.infinispan.commons.util.InfinispanCollections;
+import org.infinispan.context.Flag;
+import org.infinispan.context.InvocationContext;
+import org.infinispan.context.impl.LocalTxInvocationContext;
+import org.infinispan.context.impl.TxInvocationContext;
+import org.infinispan.factories.annotations.Inject;
+import org.infinispan.factories.annotations.Start;
+import org.infinispan.interceptors.base.BaseRpcInterceptor;
+import org.infinispan.jmx.JmxStatisticsExposer;
+import org.infinispan.jmx.annotations.DataType;
+import org.infinispan.jmx.annotations.MBean;
+import org.infinispan.jmx.annotations.ManagedAttribute;
+import org.infinispan.jmx.annotations.ManagedOperation;
+import org.infinispan.jmx.annotations.MeasurementType;
+import org.infinispan.jmx.annotations.Parameter;
+import org.infinispan.util.logging.Log;
+import org.infinispan.util.logging.LogFactory;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * This interceptor acts as a replacement to the replication interceptor when the CacheImpl is configured with
+ * ClusteredSyncMode as INVALIDATE.
+ * <p/>
+ * The idea is that rather than replicating changes to all caches in a cluster when write methods are called, simply
+ * broadcast an {@link InvalidateCommand} on the remote caches containing all keys modified.  This allows the remote
+ * cache to look up the value in a shared cache loader which would have been updated with the changes.
+ *
+ * @author Manik Surtani
+ * @author Galder Zamarreño
+ * @author Mircea.Markus@jboss.com
+ * @since 4.0
+ */
+@MBean(objectName = "Invalidation", description = "Component responsible for invalidating entries on remote caches when entries are written to locally.")
+public class TxInvalidationInterceptor extends BaseRpcInterceptor implements JmxStatisticsExposer {
+
+	private final AtomicLong invalidations = new AtomicLong( 0 );
+	private CommandsFactory commandsFactory;
+	private boolean statisticsEnabled;
+
+	private static final Log log = LogFactory.getLog( TxInvalidationInterceptor.class );
+
+	@Override
+	protected Log getLog() {
+		return log;
+	}
+
+	@Inject
+	public void injectDependencies(CommandsFactory commandsFactory) {
+		this.commandsFactory = commandsFactory;
+	}
+
+	@Start
+	private void start() {
+		this.setStatisticsEnabled( cacheConfiguration.jmxStatistics().enabled() );
+	}
+
+	@Override
+	public Object visitPutKeyValueCommand(InvocationContext ctx, PutKeyValueCommand command) throws Throwable {
+		if ( !isPutForExternalRead( command ) ) {
+			return handleInvalidate( ctx, command, command.getKey() );
+		}
+		return invokeNextInterceptor( ctx, command );
+	}
+
+	@Override
+	public Object visitReplaceCommand(InvocationContext ctx, ReplaceCommand command) throws Throwable {
+		return handleInvalidate( ctx, command, command.getKey() );
+	}
+
+	@Override
+	public Object visitRemoveCommand(InvocationContext ctx, RemoveCommand command) throws Throwable {
+		return handleInvalidate( ctx, command, command.getKey() );
+	}
+
+	@Override
+	public Object visitClearCommand(InvocationContext ctx, ClearCommand command) throws Throwable {
+		Object retval = invokeNextInterceptor( ctx, command );
+		if ( !isLocalModeForced( command ) ) {
+			// just broadcast the clear command - this is simplest!
+			if ( ctx.isOriginLocal() ) {
+				rpcManager.invokeRemotely( null, command, rpcManager.getDefaultRpcOptions( defaultSynchronous ) );
+			}
+		}
+		return retval;
+	}
+
+	@Override
+	public Object visitPutMapCommand(InvocationContext ctx, PutMapCommand command) throws Throwable {
+		Object[] keys = command.getMap() == null ? null : command.getMap().keySet().toArray();
+		return handleInvalidate( ctx, command, keys );
+	}
+
+	@Override
+	public Object visitPrepareCommand(TxInvocationContext ctx, PrepareCommand command) throws Throwable {
+		Object retval = invokeNextInterceptor( ctx, command );
+		log.tracef( "Entering InvalidationInterceptor's prepare phase.  Ctx flags are empty" );
+		// fetch the modifications before the transaction is committed (and thus removed from the txTable)
+		if ( shouldInvokeRemoteTxCommand( ctx ) ) {
+			if ( ctx.getTransaction() == null ) {
+				throw new IllegalStateException( "We must have an associated transaction" );
+			}
+
+			List<WriteCommand> mods = Arrays.asList( command.getModifications() );
+			broadcastInvalidateForPrepare( mods, ctx );
+		}
+		else {
+			log.tracef( "Nothing to invalidate - no modifications in the transaction." );
+		}
+		return retval;
+	}
+
+	@Override
+	public Object visitLockControlCommand(TxInvocationContext ctx, LockControlCommand command) throws Throwable {
+		Object retVal = invokeNextInterceptor( ctx, command );
+		if ( ctx.isOriginLocal() ) {
+			//unlock will happen async as it is a best effort
+			boolean sync = !command.isUnlock();
+			( (LocalTxInvocationContext) ctx ).remoteLocksAcquired( rpcManager.getTransport().getMembers() );
+			rpcManager.invokeRemotely( null, command, rpcManager.getDefaultRpcOptions( sync ) );
+		}
+		return retVal;
+	}
+
+	private Object handleInvalidate(InvocationContext ctx, WriteCommand command, Object... keys) throws Throwable {
+		Object retval = invokeNextInterceptor( ctx, command );
+		if ( command.isSuccessful() && !ctx.isInTxScope() ) {
+			if ( keys != null && keys.length != 0 ) {
+				if ( !isLocalModeForced( command ) ) {
+					invalidateAcrossCluster( isSynchronous( command ), keys, ctx );
+				}
+			}
+		}
+		return retval;
+	}
+
+	private void broadcastInvalidateForPrepare(List<WriteCommand> modifications, InvocationContext ctx) throws Throwable {
+		// A prepare does not carry flags, so skip checking whether is local or not
+		if ( ctx.isInTxScope() ) {
+			if ( modifications.isEmpty() ) {
+				return;
+			}
+
+			InvalidationFilterVisitor filterVisitor = new InvalidationFilterVisitor( modifications.size() );
+			filterVisitor.visitCollection( null, modifications );
+
+			if ( filterVisitor.containsPutForExternalRead ) {
+				log.debug( "Modification list contains a putForExternalRead operation.  Not invalidating." );
+			}
+			else if ( filterVisitor.containsLocalModeFlag ) {
+				log.debug( "Modification list contains a local mode flagged operation.  Not invalidating." );
+			}
+			else {
+				try {
+					invalidateAcrossCluster( defaultSynchronous, filterVisitor.result.toArray(), ctx );
+				}
+				catch (Throwable t) {
+					log.unableToRollbackEvictionsDuringPrepare( t );
+					if ( t instanceof RuntimeException ) {
+						throw (RuntimeException) t;
+					}
+					else {
+						throw new RuntimeException( "Unable to broadcast invalidation messages", t );
+					}
+				}
+			}
+		}
+	}
+
+	public static class InvalidationFilterVisitor extends AbstractVisitor {
+
+		Set<Object> result;
+		public boolean containsPutForExternalRead = false;
+		public boolean containsLocalModeFlag = false;
+
+		public InvalidationFilterVisitor(int maxSetSize) {
+			result = new HashSet<Object>( maxSetSize );
+		}
+
+		private void processCommand(FlagAffectedCommand command) {
+			containsLocalModeFlag = containsLocalModeFlag || ( command.getFlags() != null && command.getFlags().contains( Flag.CACHE_MODE_LOCAL ) );
+		}
+
+		@Override
+		public Object visitPutKeyValueCommand(InvocationContext ctx, PutKeyValueCommand command) throws Throwable {
+			processCommand( command );
+			containsPutForExternalRead =
+					containsPutForExternalRead || ( command.getFlags() != null && command.getFlags().contains( Flag.PUT_FOR_EXTERNAL_READ ) );
+			result.add( command.getKey() );
+			return null;
+		}
+
+		@Override
+		public Object visitRemoveCommand(InvocationContext ctx, RemoveCommand command) throws Throwable {
+			processCommand( command );
+			result.add( command.getKey() );
+			return null;
+		}
+
+		@Override
+		public Object visitPutMapCommand(InvocationContext ctx, PutMapCommand command) throws Throwable {
+			processCommand( command );
+			result.addAll( command.getAffectedKeys() );
+			return null;
+		}
+	}
+
+	private void invalidateAcrossCluster(boolean synchronous, Object[] keys, InvocationContext ctx) throws Throwable {
+		// increment invalidations counter if statistics maintained
+		incrementInvalidations();
+		final InvalidateCommand invalidateCommand = commandsFactory.buildInvalidateCommand( InfinispanCollections.<Flag>emptySet(), keys );
+		if ( log.isDebugEnabled() ) {
+			log.debug( "Cache [" + rpcManager.getAddress() + "] replicating " + invalidateCommand );
+		}
+
+		ReplicableCommand command = invalidateCommand;
+		if ( ctx.isInTxScope() ) {
+			TxInvocationContext txCtx = (TxInvocationContext) ctx;
+			// A Prepare command containing the invalidation command in its 'modifications' list is sent to the remote nodes
+			// so that the invalidation is executed in the same transaction and locks can be acquired and released properly.
+			// This is 1PC on purpose, as an optimisation, even if the current TX is 2PC.
+			// If the cache uses 2PC it's possible that the remotes will commit the invalidation and the originator rolls back,
+			// but this does not impact consistency and the speed benefit is worth it.
+			command = commandsFactory.buildPrepareCommand( txCtx.getGlobalTransaction(), Collections.<WriteCommand>singletonList( invalidateCommand ), true );
+		}
+		rpcManager.invokeRemotely( null, command, rpcManager.getDefaultRpcOptions( synchronous ) );
+	}
+
+	private void incrementInvalidations() {
+		if ( statisticsEnabled ) {
+			invalidations.incrementAndGet();
+		}
+	}
+
+	private boolean isPutForExternalRead(FlagAffectedCommand command) {
+		if ( command.hasFlag( Flag.PUT_FOR_EXTERNAL_READ ) ) {
+			log.trace( "Put for external read called.  Suppressing clustered invalidation." );
+			return true;
+		}
+		return false;
+	}
+
+	@ManagedOperation(
+			description = "Resets statistics gathered by this component",
+			displayName = "Reset statistics"
+	)
+	public void resetStatistics() {
+		invalidations.set( 0 );
+	}
+
+	@ManagedAttribute(
+			displayName = "Statistics enabled",
+			description = "Enables or disables the gathering of statistics by this component",
+			dataType = DataType.TRAIT,
+			writable = true
+	)
+	public boolean getStatisticsEnabled() {
+		return this.statisticsEnabled;
+	}
+
+	public void setStatisticsEnabled(
+			@Parameter(name = "enabled", description = "Whether statistics should be enabled or disabled (true/false)") boolean enabled) {
+		this.statisticsEnabled = enabled;
+	}
+
+	@ManagedAttribute(
+			description = "Number of invalidations",
+			displayName = "Number of invalidations",
+			measurementType = MeasurementType.TRENDSUP
+	)
+	public long getInvalidations() {
+		return invalidations.get();
+	}
+}

hibernate-infinispan/src/main/java/org/hibernate/cache/infinispan/access/TxPutFromLoadInterceptor.java

@@ -84,11 +84,18 @@ class TxPutFromLoadInterceptor extends BaseRpcInterceptor {
 				if (wc instanceof InvalidateCommand) {
 					// ISPN-5605 InvalidateCommand does not correctly implement getAffectedKeys()
 					for (Object key : ((InvalidateCommand) wc).getKeys()) {
+						if (log.isTraceEnabled()) {
+							log.tracef("Invalidating key %s with lock owner %s", key, ctx.getLockOwner());
+						}
 						putFromLoadValidator.beginInvalidatingKey(ctx.getLockOwner(), key);
 					}
 				}
 				else {
-					for (Object key : wc.getAffectedKeys()) {
+					Set<Object> keys = wc.getAffectedKeys();
+					if (log.isTraceEnabled()) {
+						log.tracef("Invalidating keys %s with lock owner %s", keys, ctx.getLockOwner());
+					}
+					for (Object key : keys ) {
 						putFromLoadValidator.beginInvalidatingKey(ctx.getLockOwner(), key);
 					}
 				}
@@ -99,11 +106,19 @@ class TxPutFromLoadInterceptor extends BaseRpcInterceptor {
 
 	@Override
 	public Object visitCommitCommand(TxInvocationContext ctx, CommitCommand command) throws Throwable {
+		if (log.isTraceEnabled()) {
+			log.tracef( "Commit command received, end invalidation" );
+		}
+
 		return endInvalidationAndInvokeNextInterceptor(ctx, command);
 	}
 
 	@Override
 	public Object visitRollbackCommand(TxInvocationContext ctx, RollbackCommand command) throws Throwable {
+		if (log.isTraceEnabled()) {
+			log.tracef( "Rollback command received, end invalidation" );
+		}
+
 		return endInvalidationAndInvokeNextInterceptor(ctx, command);
 	}
 
@@ -112,6 +127,11 @@ class TxPutFromLoadInterceptor extends BaseRpcInterceptor {
 			if (ctx.isOriginLocal()) {
 				// send async Commit
 				Set<Object> affectedKeys = ctx.getAffectedKeys();
+
+				if (log.isTraceEnabled()) {
+					log.tracef( "Sending end invalidation for keys %s asynchronously", affectedKeys );
+				}
+
 				if (!affectedKeys.isEmpty()) {
 					EndInvalidationCommand commitCommand = cacheCommandInitializer.buildEndInvalidationCommand(
 							cacheName, affectedKeys.toArray(), ctx.getGlobalTransaction());

hibernate-infinispan/src/test/java/org/hibernate/test/cache/infinispan/AbstractRegionAccessStrategyTest.java

@@ -47,6 +47,7 @@ import java.util.concurrent.CountDownLatch;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
@@ -334,7 +335,7 @@ public abstract class AbstractRegionAccessStrategyTest<R extends BaseRegion, S e
 
 		// Test whether the get above messes up the optimistic version
 		SessionImplementor s9 = mockedSession();
-		remoteAccessStrategy.putFromLoad(s9, KEY, VALUE1, s9.getTimestamp(), 1);
+ 		assertTrue(remoteAccessStrategy.putFromLoad(s9, KEY, VALUE1, s9.getTimestamp(), 1));
 		SessionImplementor s10 = mockedSession();
 		assertEquals(VALUE1, remoteAccessStrategy.get(s10, KEY, s10.getTimestamp()));
 		assertEquals(1, remoteRegion.getCache().size());