ISPN-7550 Remove TriangleAckInterceptor

commons/src/main/java/org/infinispan/commons/marshall/Ids.java

@@ -176,4 +176,5 @@ public interface Ids {
    int VERSIONED_RESULTS = 116;
    int FUNCTIONAL_ENTRY_VERSION_ADAPTER = 117;
 
+   int WRITE_RESPONSE = 118;
 }

commons/src/main/java/org/infinispan/commons/util/InfinispanCollections.java

@@ -157,4 +157,15 @@ public static void assertNotNullEntries(Collection<?> collection, String name) {
       Supplier<String> entrySupplier = () -> "Collection '" + name + "' contains null entry.";
       collection.forEach(k -> Objects.requireNonNull(k, entrySupplier));
    }
+
+   public static <K, V> Map<K, V> mergeMaps(Map<K, V> one, Map<K, V> second) {
+      if (one == null) {
+         return second;
+      } else if (second == null) {
+         return one;
+      } else {
+         one.putAll(second);
+         return one;
+      }
+   }
 }

core/src/main/java/org/infinispan/commands/CommandsFactory.java

@@ -58,8 +58,6 @@
 import org.infinispan.commands.write.EvictCommand;
 import org.infinispan.commands.write.ExceptionAckCommand;
 import org.infinispan.commands.write.InvalidateCommand;
-import org.infinispan.commands.write.PrimaryAckCommand;
-import org.infinispan.commands.write.PrimaryMultiKeyAckCommand;
 import org.infinispan.commands.write.PutKeyValueCommand;
 import org.infinispan.commands.write.PutMapCommand;
 import org.infinispan.commands.write.RemoveCommand;
@@ -499,12 +497,8 @@ <K, V> WriteOnlyKeyValueCommand<K, V> buildWriteOnlyKeyValueCommand(
 
    BackupAckCommand buildBackupAckCommand(long id, int topologyId);
 
-   PrimaryAckCommand buildPrimaryAckCommand();
-
    BackupMultiKeyAckCommand buildBackupMultiKeyAckCommand(long id, int segment, int topologyId);
 
-   PrimaryMultiKeyAckCommand buildPrimaryMultiKeyAckCommand(long id, int topologyId);
-
    ExceptionAckCommand buildExceptionAckCommand(long id, Throwable throwable, int topologyId);
 
    BackupWriteRcpCommand buildBackupWriteRcpCommand(DataWriteCommand command);

core/src/main/java/org/infinispan/commands/CommandsFactoryImpl.java

@@ -66,8 +66,6 @@
 import org.infinispan.commands.write.ExceptionAckCommand;
 import org.infinispan.commands.write.InvalidateCommand;
 import org.infinispan.commands.write.InvalidateL1Command;
-import org.infinispan.commands.write.PrimaryAckCommand;
-import org.infinispan.commands.write.PrimaryMultiKeyAckCommand;
 import org.infinispan.commands.write.PutKeyValueCommand;
 import org.infinispan.commands.write.PutMapCommand;
 import org.infinispan.commands.write.RemoveCommand;
@@ -531,15 +529,9 @@ public void initializeReplicableCommand(ReplicableCommand c, boolean isRemote) {
             BackupWriteRcpCommand bwc = (BackupWriteRcpCommand) c;
             bwc.init(icf, interceptorChain, notifier);
             break;
-         case PrimaryAckCommand.COMMAND_ID:
-            ((PrimaryAckCommand) c).setCommandAckCollector(commandAckCollector);
-            break;
          case BackupMultiKeyAckCommand.COMMAND_ID:
             ((BackupMultiKeyAckCommand) c).setCommandAckCollector(commandAckCollector);
             break;
-         case PrimaryMultiKeyAckCommand.COMMAND_ID:
-            ((PrimaryMultiKeyAckCommand) c).setCommandAckCollector(commandAckCollector);
-            break;
          case ExceptionAckCommand.COMMAND_ID:
             ((ExceptionAckCommand) c).setCommandAckCollector(commandAckCollector);
             break;
@@ -763,21 +755,11 @@ public BackupAckCommand buildBackupAckCommand(long id, int topologyId) {
       return new BackupAckCommand(cacheName, id, topologyId);
    }
 
-   @Override
-   public PrimaryAckCommand buildPrimaryAckCommand() {
-      return new PrimaryAckCommand(cacheName);
-   }
-
    @Override
    public BackupMultiKeyAckCommand buildBackupMultiKeyAckCommand(long id, int segment, int topologyId) {
       return new BackupMultiKeyAckCommand(cacheName, id, segment, topologyId);
    }
 
-   @Override
-   public PrimaryMultiKeyAckCommand buildPrimaryMultiKeyAckCommand(long id, int topologyId) {
-      return new PrimaryMultiKeyAckCommand(cacheName, id, topologyId);
-   }
-
    @Override
    public ExceptionAckCommand buildExceptionAckCommand(long id, Throwable throwable, int topologyId) {
       return new ExceptionAckCommand(cacheName, id, throwable, topologyId);

core/src/main/java/org/infinispan/commands/write/BackupAckCommand.java

@@ -3,12 +3,10 @@
 import java.io.IOException;
 import java.io.ObjectInput;
 import java.io.ObjectOutput;
-import java.util.concurrent.CompletableFuture;
 
 import org.infinispan.commands.remote.BaseRpcCommand;
 import org.infinispan.util.ByteString;
 import org.infinispan.util.concurrent.CommandAckCollector;
-import org.infinispan.util.concurrent.CompletableFutures;
 
 /**
  * A command that represents an acknowledge sent by a backup owner to the originator.
@@ -39,10 +37,8 @@ public BackupAckCommand(ByteString cacheName, long id, int topologyId) {
       this.topologyId = topologyId;
    }
 
-   @Override
-   public CompletableFuture<Object> invokeAsync() throws Throwable {
+   public void ack() {
       commandAckCollector.backupAck(id, getOrigin(), topologyId);
-      return CompletableFutures.completedNull();
    }
 
    @Override

core/src/main/java/org/infinispan/commands/write/BackupMultiKeyAckCommand.java

@@ -3,12 +3,10 @@
 import java.io.IOException;
 import java.io.ObjectInput;
 import java.io.ObjectOutput;
-import java.util.concurrent.CompletableFuture;
 
 import org.infinispan.commands.remote.BaseRpcCommand;
 import org.infinispan.util.ByteString;
 import org.infinispan.util.concurrent.CommandAckCollector;
-import org.infinispan.util.concurrent.CompletableFutures;
 
 /**
  * A command that represents an acknowledge sent by a backup owner to the originator.
@@ -43,10 +41,8 @@ public BackupMultiKeyAckCommand(ByteString cacheName, long id, int segment,
       this.topologyId = topologyId;
    }
 
-   @Override
-   public CompletableFuture<Object> invokeAsync() throws Throwable {
+   public void ack() {
       commandAckCollector.multiKeyBackupAck(id, getOrigin(), segment, topologyId);
-      return CompletableFutures.completedNull();
    }
 
    @Override

core/src/main/java/org/infinispan/commands/write/DataWriteCommand.java

@@ -27,15 +27,4 @@ default void initBackupWriteRcpCommand(BackupWriteRcpCommand command) {
       throw new UnsupportedOperationException();
    }
 
-   /**
-    * Initializes the primary owner acknowledge with the return value, the {@link CommandInvocationId} and the topology
-    * id.
-    *
-    * @param command          the {@link PrimaryAckCommand} to initialize.
-    * @param localReturnValue the local return value.
-    */
-   default void initPrimaryAck(PrimaryAckCommand command, Object localReturnValue) {
-      throw new UnsupportedOperationException();
-   }
-
 }

core/src/main/java/org/infinispan/commands/write/ExceptionAckCommand.java

@@ -3,12 +3,10 @@
 import java.io.IOException;
 import java.io.ObjectInput;
 import java.io.ObjectOutput;
-import java.util.concurrent.CompletableFuture;
 
 import org.infinispan.commands.remote.BaseRpcCommand;
 import org.infinispan.util.ByteString;
 import org.infinispan.util.concurrent.CommandAckCollector;
-import org.infinispan.util.concurrent.CompletableFutures;
 
 /**
  * A command that represents an exception acknowledge sent by any owner.
@@ -41,10 +39,8 @@ public ExceptionAckCommand(ByteString cacheName, long id, Throwable throwable, i
       this.topologyId = topologyId;
    }
 
-   @Override
-   public CompletableFuture<Object> invokeAsync() throws Throwable {
+   public void ack() {
       commandAckCollector.completeExceptionally(id, throwable, topologyId);
-      return CompletableFutures.completedNull();
    }
 
    @Override