ClientReauthOperation should be retryable

hazelcast-client/src/test/java/com/hazelcast/client/ClientOwnershipTest.java

@@ -18,9 +18,12 @@
 
 import com.hazelcast.client.config.ClientConfig;
 import com.hazelcast.client.impl.ClientEngineImpl;
+import com.hazelcast.client.impl.operations.ClientReAuthOperation;
 import com.hazelcast.client.test.TestHazelcastFactory;
 import com.hazelcast.core.HazelcastInstance;
 import com.hazelcast.core.IExecutorService;
+import com.hazelcast.nio.Address;
+import com.hazelcast.spi.impl.operationservice.InternalOperationService;
 import com.hazelcast.test.AssertTask;
 import com.hazelcast.test.HazelcastParallelClassRunner;
 import com.hazelcast.test.HazelcastTestSupport;
@@ -34,8 +37,11 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 import java.util.concurrent.atomic.AtomicReference;
 
+import static com.hazelcast.instance.TestUtil.getHazelcastInstanceImpl;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
@@ -109,6 +115,23 @@ public void run() throws Exception {
         });
     }
 
+    @Test
+    public void test_ClientReAuthOperation_retry() throws ExecutionException, InterruptedException {
+        HazelcastInstance instance = hazelcastFactory.newHazelcastInstance();
+        InternalOperationService operationService = getHazelcastInstanceImpl(instance).node.nodeEngine.getOperationService();
+
+        Address address = instance.getCluster().getLocalMember().getAddress();
+        ClientReAuthOperation reAuthOperation = new ClientReAuthOperation("clientUUId", 1);
+        Future<Object> future = operationService.invokeOnTarget(ClientEngineImpl.SERVICE_NAME, reAuthOperation, address);
+        future.get();
+
+        //retrying ClientReAuthOperation with same parameters, should not throw exception
+        ClientReAuthOperation reAuthOperation2 = new ClientReAuthOperation("clientUUId", 1);
+        Future<Object> future2 = operationService.invokeOnTarget(ClientEngineImpl.SERVICE_NAME, reAuthOperation2, address);
+        future2.get();
+
+    }
+
     @Test
     public void test_clientOwnedByAlreadyConnectedSecondMember_afterFirstOwnerDies() {
         final HazelcastInstance instance1 = hazelcastFactory.newHazelcastInstance();

hazelcast/src/main/java/com/hazelcast/client/impl/ClientEngineImpl.java

@@ -408,7 +408,7 @@ public boolean trySetLastAuthenticationCorrelationId(String clientUuid, long new
         AtomicLong lastCorrelationId = ConcurrencyUtil.getOrPutIfAbsent(lastAuthenticationCorrelationIds,
                 clientUuid,
                 LAST_AUTH_CORRELATION_ID_CONSTRUCTOR_FUNC);
-        return ConcurrencyUtil.setIfGreaterThan(lastCorrelationId, newCorrelationId);
+        return ConcurrencyUtil.setIfEqualOrGreaterThan(lastCorrelationId, newCorrelationId);
     }
 
     public String addOwnershipMapping(String clientUuid, String ownerUuid) {

hazelcast/src/main/java/com/hazelcast/util/ConcurrencyUtil.java

@@ -48,10 +48,10 @@ public static <E> void setMax(E obj, AtomicLongFieldUpdater<E> updater, long val
         }
     }
 
-    public static boolean setIfGreaterThan(AtomicLong oldValue, long newValue) {
+    public static boolean setIfEqualOrGreaterThan(AtomicLong oldValue, long newValue) {
         while (true) {
             long local = oldValue.get();
-            if (newValue <= local) {
+            if (newValue < local) {
                 return false;
             }
             if (oldValue.compareAndSet(local, newValue)) {

hazelcast/src/test/java/com/hazelcast/util/ConcurrencyUtilTest.java

@@ -27,9 +27,12 @@
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
 
 @RunWith(HazelcastParallelClassRunner.class)
 @Category({QuickTest.class, ParallelTest.class})
@@ -93,6 +96,13 @@ public void testGetOrPutSynchronized_whenMutexFactoryIsNull_thenThrowException()
         ConcurrencyUtil.getOrPutSynchronized(map, 5, factory, constructorFunction);
     }
 
+    @Test
+    public void testSetIfEqualOrGreaterThan() {
+        assertTrue(ConcurrencyUtil.setIfEqualOrGreaterThan(new AtomicLong(1), 1));
+        assertTrue(ConcurrencyUtil.setIfEqualOrGreaterThan(new AtomicLong(1), 2));
+        assertFalse(ConcurrencyUtil.setIfEqualOrGreaterThan(new AtomicLong(2), 1));
+    }
+
     @Test
     public void testGetOrPutIfAbsent() {
         int result = ConcurrencyUtil.getOrPutIfAbsent(map, 5, constructorFunction);