ISPN-5876 Invalidate keys on commit if 2-phase

* Invalidate keys on commit if transaction is 2-phase, otherwise
  invalidate keys on prepare.

core/src/main/java/org/infinispan/interceptors/InvalidationInterceptor.java

@@ -34,6 +34,7 @@
 import org.infinispan.commands.CommandsFactory;
 import org.infinispan.commands.FlagAffectedCommand;
 import org.infinispan.commands.control.LockControlCommand;
+import org.infinispan.commands.tx.CommitCommand;
 import org.infinispan.commands.tx.PrepareCommand;
 import org.infinispan.commands.write.ClearCommand;
 import org.infinispan.commands.write.InvalidateCommand;
@@ -134,18 +135,39 @@ public Object visitPutMapCommand(InvocationContext ctx, PutMapCommand command) t
 
    @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)) {
-         List<WriteCommand> mods = Arrays.asList(command.getModifications());
-         Transaction runningTransaction = ctx.getTransaction();
-         if (runningTransaction == null) throw new IllegalStateException("we must have an associated transaction");
-         broadcastInvalidateForPrepare(mods, runningTransaction, ctx);
-      } else {
-         log.tracef("Nothing to invalidate - no modifications in the transaction.");
+      if (command.isOnePhaseCommit()) {
+         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)) {
+            List<WriteCommand> mods = Arrays.asList(command.getModifications());
+            Transaction runningTransaction = ctx.getTransaction();
+            if (runningTransaction == null)
+               throw new IllegalStateException("we must have an associated transaction");
+            broadcastInvalidateForPrepare(mods, runningTransaction, ctx);
+         } else {
+            log.tracef("Nothing to invalidate - no modifications in the transaction.");
+         }
+         return retval;
+      }
+
+      return super.visitPrepareCommand(ctx, command);
+   }
+
+   @Override
+   public Object visitCommitCommand(TxInvocationContext ctx, CommitCommand command) throws Throwable {
+      Object retval = super.visitCommitCommand(ctx, command);
+      Set<Object> affectedKeys = ctx.getAffectedKeys();
+      try {
+         log.tracef("On commit, send invalidate for keys: %s", affectedKeys);
+         invalidateAcrossCluster(defaultSynchronous, affectedKeys.toArray());
+         return retval;
+      } catch (Throwable t) {
+         if (t instanceof RuntimeException)
+            throw t;
+         else
+            throw log.unableToBroadcastInvalidation(t);
       }
-      return retval;
    }
 
    @Override
@@ -232,7 +254,6 @@ public Object visitPutMapCommand(InvocationContext ctx, PutMapCommand command) t
       }
    }
 
-
    protected void invalidateAcrossCluster(boolean synchronous, Object[] keys) throws Throwable {
       // increment invalidations counter if statistics maintained
       incrementInvalidations();

core/src/main/java/org/infinispan/util/logging/Log.java

@@ -850,5 +850,8 @@ void asyncStoreShutdownTimeoutTooHigh(long configuredAsyncStopTimeout,
    @LogMessage(level = ERROR)
    @Message(value = "Failed to start rebalance for cache %s", id = 230)
    void rebalanceStartError(String cacheName, @Cause Throwable cause);
+
+   @Message(value = "Unable to broadcast invalidation messages", id = 231)
+   RuntimeException unableToBroadcastInvalidation(@Cause Throwable e);
 }