Merge pull request #5294 from zzhlhc/master

Support retry for WRITE_THROUGH and WRITE_BEHIND

redisson/src/main/java/org/redisson/RedissonMap.java

@@ -21,6 +21,7 @@ import org.redisson.api.*;
 import org.redisson.api.MapOptions.WriteMode;
 import org.redisson.api.listener.MapPutListener;
 import org.redisson.api.listener.MapRemoveListener;
+import org.redisson.api.map.RetryableMapWriterAsync;
 import org.redisson.api.mapreduce.RMapReduce;
 import org.redisson.client.RedisClient;
 import org.redisson.client.codec.Codec;
@@ -84,6 +85,11 @@ public class RedissonMap<K, V> extends RedissonExpirable implements RMap<K, V> {
             this.writeBehindService = null;
             writeBehindTask = null;
         }
+        if (options != null
+                && options.getWriterRetryAttempts()>1
+                && options.getWriterAsync() != null){
+            ((RetryableMapWriterAsync<Object, Object>) options.getWriterAsync()).setServiceManager(commandExecutor.getServiceManager());
+        }
     }
 
     public RedissonMap(Codec codec, CommandAsyncExecutor commandExecutor, String name) {
@@ -108,6 +114,11 @@ public class RedissonMap<K, V> extends RedissonExpirable implements RMap<K, V> {
             this.writeBehindService = null;
             writeBehindTask = null;
         }
+        if (options != null
+                && options.getWriterRetryAttempts()>1
+                && options.getWriterAsync() != null){
+            ((RetryableMapWriterAsync<Object, Object>) options.getWriterAsync()).setServiceManager(commandExecutor.getServiceManager());
+        }
     }
 
     @Override

redisson/src/main/java/org/redisson/api/MapOptions.java

@@ -19,6 +19,10 @@ import org.redisson.api.map.MapLoader;
 import org.redisson.api.map.MapLoaderAsync;
 import org.redisson.api.map.MapWriter;
 import org.redisson.api.map.MapWriterAsync;
+import org.redisson.api.map.RetryableMapWriter;
+import org.redisson.api.map.RetryableMapWriterAsync;
+
+import java.time.Duration;
 
 /**
  * Configuration for RMap object.
@@ -56,7 +60,10 @@ public class MapOptions<K, V> {
     private WriteMode writeMode = WriteMode.WRITE_THROUGH;
     private int writeBehindBatchSize = 50;
     private int writeBehindDelay = 1000;
-    
+    private int writerRetryAttempts = 0;
+    //ms
+    private long writerRetryInterval = 100;
+
     protected MapOptions() {
     }
     
@@ -89,7 +96,7 @@ public class MapOptions<K, V> {
      * @return MapOptions instance
      */
     public MapOptions<K, V> writer(MapWriter<K, V> writer) {
-        this.writer = writer;
+        this.writer = new RetryableMapWriter<>(this, writer);
         return this;
     }
     public MapWriter<K, V> getWriter() {
@@ -103,7 +110,7 @@ public class MapOptions<K, V> {
      * @return MapOptions instance
      */
     public MapOptions<K, V> writerAsync(MapWriterAsync<K, V> writer) {
-        this.writerAsync = writer;
+        this.writerAsync = new RetryableMapWriterAsync<>(this, writer);
         return this;
     }
     public MapWriterAsync<K, V> getWriterAsync() {
@@ -159,7 +166,43 @@ public class MapOptions<K, V> {
     public WriteMode getWriteMode() {
         return writeMode;
     }
-    
+
+    public int getWriterRetryAttempts() {
+        return writerRetryAttempts;
+    }
+
+    /**
+     * Sets max retry attempts for {@link RetryableMapWriter} or {@link RetryableMapWriterAsync}
+     *
+     * @param writerRetryAttempts object
+     * @return MapOptions instance
+     */
+    public MapOptions<K, V> writerRetryAttempts(int writerRetryAttempts) {
+        if (writerRetryAttempts <= 0){
+            throw new IllegalArgumentException("writerRetryAttempts must be bigger than 0");
+        }
+        this.writerRetryAttempts = writerRetryAttempts;
+        return this;
+    }
+
+    public long getWriterRetryInterval() {
+        return writerRetryInterval;
+    }
+
+    /**
+     * Sets retry interval for {@link RetryableMapWriter} or {@link RetryableMapWriterAsync}
+     * 
+     * @param writerRetryInterval {@link Duration}
+     * @return MapOptions instance
+     */
+    public MapOptions<K, V> writerRetryInterval(Duration writerRetryInterval) {
+        if (writerRetryInterval.isNegative()) {
+            throw new IllegalArgumentException("writerRetryInterval must be positive");
+        }
+        this.writerRetryInterval = writerRetryInterval.toMillis();
+        return this;
+    }
+
     /**
      * Sets {@link MapLoader} object.
      * 

redisson/src/main/java/org/redisson/api/map/RetryableMapWriter.java

@@ -0,0 +1,82 @@
+/**
+ * Copyright (c) 2013-2022 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.redisson.api.map;
+
+import org.redisson.api.MapOptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.Map;
+
+public class RetryableMapWriter<K, V> implements MapWriter<K, V> {
+
+    private static final Logger log = LoggerFactory.getLogger(RetryableMapWriter.class);
+
+    private final MapOptions<K, V> options;
+    private final MapWriter<K, V> mapWriter;
+
+    public RetryableMapWriter(MapOptions<K, V> options, MapWriter<K, V> mapWriter) {
+        this.options = options;
+        this.mapWriter = mapWriter;
+    }
+
+    @Override
+    public void write(Map<K, V> addedMap) {
+        //execute at least once
+        int leftAddAttempts = Math.max(1, options.getWriterRetryAttempts());
+        while (leftAddAttempts > 0) {
+            try {
+                //do write
+                mapWriter.write(addedMap);
+                break;
+            } catch (Exception exception) {
+                if (--leftAddAttempts == 0) {
+                    throw exception;
+                } else {
+                    log.warn("Unable to add keys: {}, will retry after {}ms", addedMap, options.getWriterRetryInterval(), exception);
+                    try {
+                        Thread.sleep(options.getWriterRetryInterval());
+                    } catch (InterruptedException ignore) {
+                    }
+                }
+            }
+        }
+    }
+
+    @Override
+    public void delete(Collection<K> keys) {
+        //execute at least once
+        int leftDeleteAttempts = Math.max(1, options.getWriterRetryAttempts());
+        while (leftDeleteAttempts > 0) {
+            try {
+                //do delete
+                mapWriter.delete(keys);
+                break;
+            } catch (Exception exception) {
+                if (--leftDeleteAttempts == 0) {
+                    throw exception;
+                } else {
+                    log.warn("Unable to delete keys: {}, will retry after {}ms", keys, options.getWriterRetryInterval(), exception);
+                    try {
+                        Thread.sleep(options.getWriterRetryInterval());
+                    } catch (InterruptedException ignore) {
+                    }
+                }
+            }
+        }
+    }
+}

redisson/src/main/java/org/redisson/api/map/RetryableMapWriterAsync.java

@@ -0,0 +1,113 @@
+/**
+ * Copyright (c) 2013-2022 Nikita Koksharov
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.redisson.api.map;
+
+import org.redisson.api.MapOptions;
+import org.redisson.connection.ServiceManager;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Collection;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CompletionStage;
+import java.util.concurrent.TimeUnit;
+
+public class RetryableMapWriterAsync<K, V> implements MapWriterAsync<K, V> {
+
+    private static final Logger log = LoggerFactory.getLogger(RetryableMapWriterAsync.class);
+
+    private final MapOptions<K, V> options;
+
+    private final MapWriterAsync<K, V> mapWriterAsync;
+
+    private ServiceManager serviceManager;
+
+    public void setServiceManager(ServiceManager serviceManager) {
+        this.serviceManager = serviceManager;
+    }
+
+    public RetryableMapWriterAsync(MapOptions<K, V> options, MapWriterAsync<K, V> mapWriterAsync) {
+        this.options = options;
+        this.mapWriterAsync = mapWriterAsync;
+    }
+
+    @Override
+    public CompletionStage<Void> write(Map<K, V> addedMap) {
+        CompletableFuture<Void> result = new CompletableFuture<>();
+        retryWrite(Math.max(1, options.getWriterRetryAttempts()), addedMap, result);
+        return result;
+    }
+
+    private void retryWrite(int leftAttempts, Map<K, V> addedMap, CompletableFuture<Void> result) {
+        mapWriterAsync.write(addedMap).whenComplete((x, e) -> {
+                    if (e == null) {
+                        result.complete(null);
+                        return;
+                    }
+
+                    if (leftAttempts - 1 <= 0) {
+                        result.completeExceptionally(e);
+                        return;
+                    }
+
+                    if (serviceManager == null) {
+                        log.warn("The serviceManager is null, so cannot retry writing keys: {}", addedMap);
+                        result.completeExceptionally(e);
+                        return;
+                    }
+
+                    log.warn("Unable to add keys: {}, will retry after {}ms", addedMap, options.getWriterRetryInterval(), e);
+                    serviceManager.newTimeout(t -> retryWrite(leftAttempts - 1, addedMap, result),
+                            options.getWriterRetryInterval(), TimeUnit.MILLISECONDS);
+                }
+        );
+    }
+
+    @Override
+    public CompletionStage<Void> delete(Collection<K> keys) {
+        CompletableFuture<Void> result = new CompletableFuture<>();
+        retryDelete(Math.max(1, options.getWriterRetryAttempts()), keys, result);
+        return result;
+    }
+
+    private void retryDelete(int leftAttempts, Collection<K> keys, CompletableFuture<Void> result) {
+        mapWriterAsync.delete(keys).whenComplete((x, e) -> {
+                    if (e == null) {
+                        result.complete(null);
+                        return;
+                    }
+
+                    if (leftAttempts - 1 <= 0) {
+                        result.completeExceptionally(e);
+                        return;
+                    }
+
+                    if (serviceManager == null) {
+                        log.warn("The serviceManager is null so cannot retry deleting keys: {}", keys);
+                        result.completeExceptionally(e);
+                        return;
+                    }
+
+                    log.warn("Unable to delete keys: {}, will retry after {}ms", keys, options.getWriterRetryInterval(), e);
+                    serviceManager.newTimeout(t -> retryDelete(leftAttempts - 1, keys, result),
+                            options.getWriterRetryInterval(), TimeUnit.MILLISECONDS);
+                }
+        );
+    }
+
+
+}

redisson/src/test/java/org/redisson/BaseMapTest.java

@@ -24,6 +24,8 @@ import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -1438,7 +1440,124 @@ public abstract class BaseMapTest extends BaseTest {
         assertThat(store).isEqualTo(expected);
         destroy(map);
     }
-    
+
+    @Test
+    public void testRetryableWriterAsyncSuccessAtLastRetry() throws InterruptedException {
+        //success at last retry
+        int expectedRetryAttempts = 3;
+        AtomicInteger actualRetryTimes = new AtomicInteger(0);
+        Map<String, String> store = new HashMap<>();
+        MapOptions<String, String> options = MapOptions.<String, String>defaults()
+                .writerRetryAttempts(expectedRetryAttempts)
+                .writerAsync(new MapWriterAsync<String, String>() {
+                    @Override
+                    public CompletionStage<Void> write(Map<String, String> map) {
+                        return CompletableFuture.runAsync(()->{
+                            //throws until last chance
+                            if (actualRetryTimes.incrementAndGet() < expectedRetryAttempts) {
+                                throw new IllegalStateException("retry");
+                            }
+                            store.putAll(map);
+                        });
+                    }
+
+                    @Override
+                    public CompletionStage<Void> delete(Collection<String> keys) {
+                        return CompletableFuture.runAsync(()->{
+                            if (actualRetryTimes.incrementAndGet() < expectedRetryAttempts) {
+                                throw new IllegalStateException("retry");
+                            }
+                            keys.forEach(store::remove);
+                        });
+                    }
+                })
+                .writeMode(MapOptions.WriteMode.WRITE_BEHIND)
+                .writerRetryInterval(Duration.ofMillis(100));
+
+        final RMap<String, String> map = redisson.getMap("test", options);
+        //do add
+        map.put("1", "11");
+        Thread.sleep(2400);
+        
+        //assert add
+        Map<String, String> expectedMap = new HashMap<>();
+        expectedMap.put("1", "11");
+        assertThat(store).isEqualTo(expectedMap);
+
+        //assert add retry times
+        assertThat(actualRetryTimes.get()).isEqualTo(expectedRetryAttempts);
+
+        //do delete
+        actualRetryTimes.set(0);
+        map.remove("1");
+        Thread.sleep(2400);
+
+        //assert delete 
+        expectedMap.clear();
+        assertThat(store).isEqualTo(expectedMap);
+
+        //assert delete retry times
+        assertThat(actualRetryTimes.get()).isEqualTo(expectedRetryAttempts);
+        destroy(map);
+    }
+
+    @Test
+    public void testRetryableWriterSuccessAtLastRetry() throws InterruptedException {
+        //success at last retry
+        int expectedRetryAttempts = 3;
+        AtomicInteger actualRetryTimes = new AtomicInteger(0);
+        Map<String, String> store = new HashMap<>();
+        MapOptions<String, String> options = MapOptions.<String, String>defaults()
+                .writerRetryAttempts(expectedRetryAttempts)
+                .writer(new MapWriter<String, String>() {
+                    @Override
+                    public void write(Map<String, String> map) {
+                        if (actualRetryTimes.incrementAndGet() < expectedRetryAttempts) {
+                            throw new IllegalStateException("retry");
+                        }
+                        store.putAll(map);
+                    }
+
+                    @Override
+                    public void delete(Collection<String> keys) {
+                        if (actualRetryTimes.incrementAndGet() < expectedRetryAttempts) {
+                            throw new IllegalStateException("retry");
+                        }
+                        keys.forEach(store::remove);
+                    }
+                })
+                .writeMode(MapOptions.WriteMode.WRITE_THROUGH);
+
+        final RMap<String, String> map = redisson.getMap("test", options);
+        
+        //do add
+        map.put("1", "11");
+        Thread.sleep(1400);
+
+        //assert add
+        Map<String, String> expectedMap = new HashMap<>();
+        expectedMap.put("1", "11");
+        assertThat(store).isEqualTo(expectedMap);
+
+        //assert add retry times
+        assertThat(actualRetryTimes.get()).isEqualTo(expectedRetryAttempts);
+        
+        
+        //do delete
+        actualRetryTimes.set(0);
+        map.remove("1");
+        Thread.sleep(1400);
+        
+        //assert delete 
+        expectedMap.clear();
+        assertThat(store).isEqualTo(expectedMap);
+
+        //assert delete retry times
+        assertThat(actualRetryTimes.get()).isEqualTo(expectedRetryAttempts);
+        destroy(map);
+    }
+
+
     @Test
     public void testLoadAllReplaceValues() {
         Map<String, String> cache = new HashMap<>();