ISPN-5046 PartitionHandling: split during commit can leave the cache …

…inconsistent after merge

* keep the transaction resources while the partition is not healed.
* when the merge happens, complete the pending transactions.

core/src/main/java/org/infinispan/cache/impl/CacheImpl.java

@@ -948,11 +948,7 @@ public String getCacheStatus() {
 
    @Override
    public AvailabilityMode getAvailability() {
-      if (partitionHandlingManager != null) {
-         return partitionHandlingManager.getAvailabilityMode();
-      } else {
-         return AvailabilityMode.AVAILABLE;
-      }
+      return partitionHandlingManager.getAvailabilityMode();
    }
 
    @Override
@@ -1071,7 +1067,7 @@ public Stats getStats() {
 
    @Override
    public XAResource getXAResource() {
-      return new TransactionXaAdapter(txTable, recoveryManager, txCoordinator, commandsFactory, rpcManager, null, config, name);
+      return new TransactionXaAdapter(txTable, recoveryManager, txCoordinator, commandsFactory, rpcManager, null, config, name, partitionHandlingManager);
    }
 
    @Override

core/src/main/java/org/infinispan/commands/remote/recovery/TxCompletionNotificationCommand.java

@@ -114,7 +114,7 @@ public GlobalTransaction getGlobalTransaction() {
    private void forwardCommandRemotely(RemoteTransaction remoteTx) {
       Set<Object> affectedKeys = remoteTx.getAffectedKeys();
       log.tracef("Invoking forward of TxCompletionNotification for transaction %s. Affected keys: %s", gtx, affectedKeys);
-      stateTransferManager.forwardCommandIfNeeded(this, affectedKeys, remoteTx.getGlobalTransaction().getAddress(), false);
+      stateTransferManager.forwardCommandIfNeeded(this, affectedKeys, remoteTx.getGlobalTransaction().getAddress());
    }
 
    @Override

core/src/main/java/org/infinispan/factories/PartitionHandlingManagerFactory.java

@@ -2,6 +2,7 @@
 
 
 import org.infinispan.factories.annotations.DefaultFactoryFor;
+import org.infinispan.partitionhandling.impl.AvailablePartitionHandlingManager;
 import org.infinispan.partitionhandling.impl.PartitionHandlingManager;
 import org.infinispan.partitionhandling.impl.PartitionHandlingManagerImpl;
 
@@ -10,8 +11,8 @@
  * @since 7.0
  */
 @DefaultFactoryFor(classes = PartitionHandlingManager.class)
-public class PartitionHandlingManagerFactory extends AbstractNamedCacheComponentFactory implements
-      AutoInstantiableFactory {
+public class PartitionHandlingManagerFactory extends AbstractNamedCacheComponentFactory implements AutoInstantiableFactory {
+
    @Override
    @SuppressWarnings("unchecked")
    public <T> T construct(Class<T> componentType) {
@@ -21,6 +22,6 @@ public <T> T construct(Class<T> componentType) {
             return (T) new PartitionHandlingManagerImpl();
          }
       }
-      return null;
+      return (T) AvailablePartitionHandlingManager.getInstance();
    }
 }

core/src/main/java/org/infinispan/interceptors/distribution/BaseDistributionInterceptor.java

@@ -1,7 +1,6 @@
 package org.infinispan.interceptors.distribution;
 
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -267,11 +266,9 @@ protected final Object handleNonTxWriteCommand(InvocationContext ctx, DataWriteC
          throw new CacheException("Attempted execution of non-transactional write command in a transactional invocation context");
       }
 
-      RecipientGenerator recipientGenerator = new SingleKeyRecipientGenerator(command.getKey());
-
       // see if we need to load values from remote sources first
       if (needValuesFromPreviousOwners(ctx, command)) {
-         remoteGetBeforeWrite(ctx, command, recipientGenerator);
+         remoteGetBeforeWrite(ctx, command, command.getKey());
       }
 
       // invoke the command locally, we need to know if it's successful or not
@@ -309,7 +306,7 @@ protected final Object handleNonTxWriteCommand(InvocationContext ctx, DataWriteC
                log.tracef("Skipping the replication of the conditional command as it did not succeed on primary owner (%s).", command);
                return localResult;
             }
-            List<Address> recipients = recipientGenerator.generateRecipients();
+            List<Address> recipients = cdl.getOwners(command.getKey());
             // Ignore the previous value on the backup owners
             command.setValueMatcher(ValueMatcher.MATCH_ALWAYS);
             try {
@@ -327,7 +324,7 @@ protected final Object handleNonTxWriteCommand(InvocationContext ctx, DataWriteC
                log.tracef("Skipping the replication of the command as it did not succeed on primary owner (%s).", command);
                return localResult;
             }
-            List<Address> recipients = recipientGenerator.generateRecipients();
+            List<Address> recipients = cdl.getOwners(command.getKey());
             log.tracef("I'm the primary owner, sending the command to all the backups (%s) in order to be applied.",
                   recipients);
             // check if a single owner has been configured and the target for the key is the local address
@@ -517,59 +514,5 @@ public Object visitGetAllCommand(InvocationContext ctx, GetAllCommand command) t
     */
    protected abstract boolean needValuesFromPreviousOwners(InvocationContext ctx, WriteCommand command);
 
-   protected abstract void remoteGetBeforeWrite(InvocationContext ctx, WriteCommand command, RecipientGenerator keygen) throws Throwable;
-
-   interface RecipientGenerator {
-
-      Collection<Object> getKeys();
-
-      List<Address> generateRecipients();
-   }
-
-   class SingleKeyRecipientGenerator implements RecipientGenerator {
-      private final Object key;
-      private final Set<Object> keys;
-      private List<Address> recipients = null;
-
-      SingleKeyRecipientGenerator(Object key) {
-         this.key = key;
-         keys = Collections.singleton(key);
-      }
-
-      @Override
-      public List<Address> generateRecipients() {
-         if (recipients == null) {
-            recipients = cdl.getOwners(key);
-         }
-         return recipients;
-      }
-
-      @Override
-      public Collection<Object> getKeys() {
-         return keys;
-      }
-   }
-
-   class MultipleKeysRecipientGenerator implements RecipientGenerator {
-
-      private final Collection<Object> keys;
-      private List<Address> recipients = null;
-
-      MultipleKeysRecipientGenerator(Collection<Object> keys) {
-         this.keys = keys;
-      }
-
-      @Override
-      public List<Address> generateRecipients() {
-         if (recipients == null) {
-            recipients = cdl.getOwners(keys);
-         }
-         return recipients;
-      }
-
-      @Override
-      public Collection<Object> getKeys() {
-         return keys;
-      }
-   }
+   protected abstract void remoteGetBeforeWrite(InvocationContext ctx, WriteCommand command, Object key) throws Throwable;
 }

core/src/main/java/org/infinispan/interceptors/distribution/NonTxDistributionInterceptor.java

@@ -204,13 +204,11 @@ public Object visitReplaceCommand(InvocationContext ctx, ReplaceCommand command)
       return handleNonTxWriteCommand(ctx, command);
    }
 
-   protected void remoteGetBeforeWrite(InvocationContext ctx, WriteCommand command, RecipientGenerator keygen) throws Throwable {
-      for (Object k : keygen.getKeys()) {
-         if (cdl.localNodeIsPrimaryOwner(k)) {
-            // Then it makes sense to try a local get and wrap again. This will compensate the fact the the entry was not local
-            // earlier when the EntryWrappingInterceptor executed during current invocation context but it should be now.
-            localGetCacheEntry(ctx, k, true, command);
-         }
+   protected void remoteGetBeforeWrite(InvocationContext ctx, WriteCommand command, Object key) throws Throwable {
+      if (cdl.localNodeIsPrimaryOwner(key)) {
+         // Then it makes sense to try a local get and wrap again. This will compensate the fact the the entry was not local
+         // earlier when the EntryWrappingInterceptor executed during current invocation context but it should be now.
+         localGetCacheEntry(ctx, key, true, command);
       }
    }