Remove ICache wan operation provider.

It was used to check wan queues on caller side but it is wrong place
to check queues with this commit the check is moved to callee side.

hazelcast/src/main/java/com/hazelcast/cache/impl/CacheService.java

@@ -157,6 +157,11 @@ public CacheWanEventPublisher getCacheWanEventPublisher() {
         throw new UnsupportedOperationException("WAN replication is not supported");
     }
 
+    @Override
+    public void checkWanReplicationQueues(String cacheName) {
+        // NOP intentionally
+    }
+
     public static ObjectNamespace getObjectNamespace(String cacheName) {
         return new DistributedObjectNamespace(SERVICE_NAME, cacheName);
     }

hazelcast/src/main/java/com/hazelcast/cache/impl/ICacheService.java

@@ -25,10 +25,10 @@
 import com.hazelcast.nio.serialization.Data;
 import com.hazelcast.spi.EventFilter;
 import com.hazelcast.spi.EventPublishingService;
-import com.hazelcast.spi.partition.FragmentedMigrationAwareService;
 import com.hazelcast.spi.ManagedService;
 import com.hazelcast.spi.NodeEngine;
 import com.hazelcast.spi.RemoteService;
+import com.hazelcast.spi.partition.FragmentedMigrationAwareService;
 
 import java.util.Collection;
 
@@ -135,6 +135,13 @@ public interface ICacheService
      */
     CacheWanEventPublisher getCacheWanEventPublisher();
 
+
+    /**
+     * @param cacheNameWithPrefix the full name of the {@link
+     *                            com.hazelcast.cache.ICache}, including the manager scope prefix
+     */
+    void checkWanReplicationQueues(String cacheNameWithPrefix);
+
     /**
      * Returns an interface for interacting with the cache event journals.
      */
@@ -145,9 +152,9 @@ public interface ICacheService
      * cluster version 3.10 or greater, the cluster-wide invocation ensures that all members of
      * the cluster will receive the cache config even in the face of cluster membership changes.
      *
-     * @param cacheConfig   the cache config to create on all members of the cluster
-     * @param <K>           key type parameter
-     * @param <V>           value type parameter
+     * @param cacheConfig the cache config to create on all members of the cluster
+     * @param <K>         key type parameter
+     * @param <V>         value type parameter
      * @since 3.10
      */
     <K, V> void createCacheConfigOnAllMembers(PreJoinCacheConfig<K, V> cacheConfig);

hazelcast/src/main/java/com/hazelcast/cache/impl/operation/CacheOperation.java

@@ -98,6 +98,7 @@ public final void beforeRun() throws Exception {
 
         if (recordStore != null && recordStore.isWanReplicationEnabled()) {
             wanEventPublisher = cacheService.getCacheWanEventPublisher();
+            cacheService.checkWanReplicationQueues(name);
         }
 
         beforeRunInternal();