introduced queueing service for operations unsent to master

Change-Id: I0021082ec96b26b4302821a37acf2eb085baf2af

java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/RegattaLogFixTrackerRegattaListener.java

@@ -27,8 +27,11 @@ import com.sap.sailing.server.interfaces.RacingEventService;
 import com.sap.sse.replication.OperationExecutionListener;
 import com.sap.sse.replication.OperationWithResult;
 import com.sap.sse.replication.OperationWithResultWithIdWrapper;
+import com.sap.sse.replication.OperationsToMasterSender;
 import com.sap.sse.replication.ReplicableWithObjectInputStream;
 import com.sap.sse.replication.ReplicationMasterDescriptor;
+import com.sap.sse.replication.ReplicationService;
+import com.sap.sse.replication.UnsentOperationsToMasterSender;
 
 /**
  * This is the main entry point of the {@link SensorFixStore} based fix tracking.
@@ -51,6 +54,14 @@ public class RegattaLogFixTrackerRegattaListener extends AbstractTrackedRegattaA
     private final ServiceTracker<RacingEventService, RacingEventService> racingEventServiceTracker;
     private final SensorFixMapperFactory sensorFixMapperFactory;
 
+    /**
+     * This field is expected to be set by the {@link ReplicationService} once it has "adopted" this replicable.
+     * The {@link ReplicationService} "injects" this service so it can be used here as a delegate for the
+     * {@link UnsentOperationsToMasterSender#retrySendingLater(OperationWithResult, OperationsToMasterSender)}
+     * method.
+     */
+    private UnsentOperationsToMasterSender unsentOperationsToMasterSender;
+
     public RegattaLogFixTrackerRegattaListener(
             ServiceTracker<RacingEventService, RacingEventService> racingEventServiceTracker,
             SensorFixMapperFactory sensorFixMapperFactory) {
@@ -215,4 +226,17 @@ public class RegattaLogFixTrackerRegattaListener extends AbstractTrackedRegattaA
     private boolean isReplica() {
         return master != null;
     }
+
+    @Override
+    public void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service) {
+        this.unsentOperationsToMasterSender = service;
+    }
+
+    @Override
+    public <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
+            OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
+        if (unsentOperationsToMasterSender != null) {
+            unsentOperationsToMasterSender.retrySendingLater(operationWithResult, sender);
+        }
+    }
 }

java/com.sap.sailing.polars/src/com/sap/sailing/polars/impl/PolarDataServiceImpl.java

@@ -52,7 +52,10 @@ import com.sap.sse.datamining.shared.GroupKey;
 import com.sap.sse.replication.OperationExecutionListener;
 import com.sap.sse.replication.OperationWithResult;
 import com.sap.sse.replication.OperationWithResultWithIdWrapper;
+import com.sap.sse.replication.OperationsToMasterSender;
 import com.sap.sse.replication.ReplicationMasterDescriptor;
+import com.sap.sse.replication.ReplicationService;
+import com.sap.sse.replication.UnsentOperationsToMasterSender;
 import com.sap.sse.util.ClearStateTestSupport;
 
 /**
@@ -86,6 +89,14 @@ public class PolarDataServiceImpl implements ReplicablePolarService, ClearStateT
 
     private DomainFactory domainFactory;
 
+    /**
+     * This field is expected to be set by the {@link ReplicationService} once it has "adopted" this replicable.
+     * The {@link ReplicationService} "injects" this service so it can be used here as a delegate for the
+     * {@link UnsentOperationsToMasterSender#retrySendingLater(OperationWithResult, OperationsToMasterSender)}
+     * method.
+     */
+    private UnsentOperationsToMasterSender unsentOperationsToMasterSender;
+
     /**
      * Constructs the polar data service with default generation settings.
      */
@@ -421,4 +432,17 @@ public class PolarDataServiceImpl implements ReplicablePolarService, ClearStateT
     public void clearState() throws Exception {
         resetState();
     }
+
+    @Override
+    public void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service) {
+        this.unsentOperationsToMasterSender = service;
+    }
+
+    @Override
+    public <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
+            OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
+        if (unsentOperationsToMasterSender != null) {
+            unsentOperationsToMasterSender.retrySendingLater(operationWithResult, sender);
+        }
+    }
 }

java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/UnsentOperationSenderJobTest.java

@@ -0,0 +1,74 @@
+package com.sap.sailing.server.replication.test;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.io.Serializable;
+import java.util.concurrent.BrokenBarrierException;
+import java.util.concurrent.CountDownLatch;
+
+import org.junit.Before;
+import org.junit.Test;
+
+import com.sap.sailing.server.interfaces.RacingEventService;
+import com.sap.sse.replication.OperationWithResult;
+import com.sap.sse.replication.OperationWithResultWithIdWrapper;
+import com.sap.sse.replication.OperationsToMasterSender;
+import com.sap.sse.replication.ReplicationMasterDescriptor;
+import com.sap.sse.replication.impl.UnsentOperationsSenderJob;
+
+public class UnsentOperationSenderJobTest implements OperationsToMasterSender<RacingEventService, OperationWithResult<RacingEventService, String>> {
+    private UnsentOperationsSenderJob job;
+    private int resendCount;
+    private CountDownLatch latch;
+    
+//    @Rule public Timeout timeout = new Timeout(10, TimeUnit.SECONDS);
+    
+    @Before
+    public void setUp() {
+        job = new UnsentOperationsSenderJob();
+        resendCount = 0;
+        latch = new CountDownLatch(1);
+    }
+    
+    @Test
+    public void testResendThatFailsAtLeastOnce() throws InterruptedException, BrokenBarrierException {
+        final OperationWithResult<RacingEventService, String> operationWithResult = null;
+        final OperationsToMasterSender<RacingEventService, OperationWithResult<RacingEventService, String>> sender = this;
+        job.retrySendingLater(operationWithResult, sender);
+        latch.await();
+        assertEquals(2, resendCount);
+    }
+
+    public <T> void sendReplicaInitiatedOperationToMaster(OperationWithResult<RacingEventService, T> operation) throws IOException {
+        logger.info("resending at count "+resendCount);
+        if (resendCount++ == 0) {
+            throw new IOException("First resend failed");
+        } else {
+            latch.countDown();
+        }
+    }
+    
+    @Override
+    public <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
+            OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
+        job.retrySendingLater(operationWithResult, sender);
+    }
+
+    @Override
+    public Serializable getId() {
+        return getClass().getName();
+    }
+
+    @Override
+    public ReplicationMasterDescriptor getMasterDescriptor() {
+        return null;
+    }
+
+    @Override
+    public void writeOperation(OperationWithResult<?, ?> operation, OutputStream outputStream, boolean closeStream) throws IOException {}
+
+    @Override
+    public void addOperationSentToMasterForReplication(OperationWithResultWithIdWrapper<RacingEventService, ?> operationWithResultWithIdWrapper) {}
+}

java/com.sap.sailing.server/src/com/sap/sailing/server/impl/RacingEventServiceImpl.java

@@ -285,7 +285,10 @@ import com.sap.sse.pairinglist.PairingListTemplateFactory;
 import com.sap.sse.replication.OperationExecutionListener;
 import com.sap.sse.replication.OperationWithResult;
 import com.sap.sse.replication.OperationWithResultWithIdWrapper;
+import com.sap.sse.replication.OperationsToMasterSender;
 import com.sap.sse.replication.ReplicationMasterDescriptor;
+import com.sap.sse.replication.ReplicationService;
+import com.sap.sse.replication.UnsentOperationsToMasterSender;
 import com.sap.sse.shared.media.ImageDescriptor;
 import com.sap.sse.shared.media.VideoDescriptor;
 import com.sap.sse.util.ClearStateTestSupport;
@@ -510,6 +513,14 @@ public class RacingEventServiceImpl implements RacingEventService, ClearStateTes
 
     private final PairingListTemplateFactory pairingListTemplateFactory = PairingListTemplateFactory.INSTANCE; 
     
+    /**
+     * This field is expected to be set by the {@link ReplicationService} once it has "adopted" this replicable.
+     * The {@link ReplicationService} "injects" this service so it can be used here as a delegate for the
+     * {@link UnsentOperationsToMasterSender#retrySendingLater(OperationWithResult, OperationsToMasterSender)}
+     * method.
+     */
+    private UnsentOperationsToMasterSender unsentOperationsToMasterSender;
+
     /**
      * Providing the constructor parameters for a new {@link RacingEventServiceImpl} instance is a bit tricky
      * in some cases because containment and initialization order of some types is fairly tightly coupled.
@@ -4352,4 +4363,17 @@ public class RacingEventServiceImpl implements RacingEventService, ClearStateTes
                 competitorDescriptor.getTimeOnDistanceAllowancePerNauticalMile(), searchTag, boat);
         return competitorWithBoat;
     }
+
+    @Override
+    public void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service) {
+        this.unsentOperationsToMasterSender = service;
+    }
+
+    @Override
+    public <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
+            OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
+        if (unsentOperationsToMasterSender != null) {
+            unsentOperationsToMasterSender.retrySendingLater(operationWithResult, sender);
+        }
+    }
 }

java/com.sap.sse.filestorage/src/com/sap/sse/filestorage/impl/FileStorageManagementServiceImpl.java

@@ -26,7 +26,10 @@ import com.sap.sse.filestorage.FileStorageServiceResolver;
 import com.sap.sse.replication.OperationExecutionListener;
 import com.sap.sse.replication.OperationWithResult;
 import com.sap.sse.replication.OperationWithResultWithIdWrapper;
+import com.sap.sse.replication.OperationsToMasterSender;
 import com.sap.sse.replication.ReplicationMasterDescriptor;
+import com.sap.sse.replication.ReplicationService;
+import com.sap.sse.replication.UnsentOperationsToMasterSender;
 
 /**
  * Implements {@link ServiceTrackerCustomizer} so that all {@link FileStorageServices} announced in the registry can
@@ -55,6 +58,14 @@ public class FileStorageManagementServiceImpl implements ReplicableFileStorageMa
     private FileStorageServicePropertyStore propertyStore;
     private final FileStorageServiceResolver serviceResolver;
 
+    /**
+     * This field is expected to be set by the {@link ReplicationService} once it has "adopted" this replicable.
+     * The {@link ReplicationService} "injects" this service so it can be used here as a delegate for the
+     * {@link UnsentOperationsToMasterSender#retrySendingLater(OperationWithResult, OperationsToMasterSender)}
+     * method.
+     */
+    private UnsentOperationsToMasterSender unsentOperationForMasterQueue;
+
     public FileStorageManagementServiceImpl(TypeBasedServiceFinder<FileStorageService> serviceFinder,
             FileStorageServicePropertyStore propertyStore) {
         this.serviceFinder = serviceFinder;
@@ -235,4 +246,17 @@ public class FileStorageManagementServiceImpl implements ReplicableFileStorageMa
     public void setCurrentlyApplyingOperationReceivedFromMaster(boolean currentlyApplyingOperationReceivedFromMaster) {
         this.currentlyApplyingOperationReceivedFromMaster.set(currentlyApplyingOperationReceivedFromMaster);
     }
+
+    @Override
+    public void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service) {
+        this.unsentOperationForMasterQueue = service;
+    }
+
+    @Override
+    public <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
+            OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
+        if (unsentOperationForMasterQueue != null) {
+            unsentOperationForMasterQueue.retrySendingLater(operationWithResult, sender);
+        }
+    }
 }

java/com.sap.sse.mail/src/com/sap/sse/mail/impl/MailServiceImpl.java

@@ -31,7 +31,9 @@ import com.sap.sse.mail.SerializableMultipartSupplier;
 import com.sap.sse.replication.OperationExecutionListener;
 import com.sap.sse.replication.OperationWithResult;
 import com.sap.sse.replication.OperationWithResultWithIdWrapper;
+import com.sap.sse.replication.OperationsToMasterSender;
 import com.sap.sse.replication.ReplicationMasterDescriptor;
+import com.sap.sse.replication.UnsentOperationsToMasterSender;
 import com.sap.sse.util.ObjectInputStreamResolvingAgainstCache;
 
 public class MailServiceImpl implements ReplicableMailService {
@@ -51,6 +53,14 @@ public class MailServiceImpl implements ReplicableMailService {
     private ThreadLocal<Boolean> currentlyApplyingOperationReceivedFromMaster = ThreadLocal.withInitial(() -> false);
 
     private final MailServiceResolver mailServiceResolver;
+    
+    /**
+     * This field is expected to be set by the {@link ReplicationService} once it has "adopted" this replicable.
+     * The {@link ReplicationService} "injects" this service so it can be used here as a delegate for the
+     * {@link UnsentOperationsToMasterSender#retrySendingLater(OperationWithResult, OperationsToMasterSender)}
+     * method.
+     */
+    private UnsentOperationsToMasterSender unsentOperationForMasterQueue;
 
     public MailServiceImpl(Properties mailProperties, MailServiceResolver mailServiceResolver) {
         this.mailProperties = mailProperties;
@@ -244,4 +254,17 @@ public class MailServiceImpl implements ReplicableMailService {
     public void setCurrentlyApplyingOperationReceivedFromMaster(boolean currentlyApplyingOperationReceivedFromMaster) {
         this.currentlyApplyingOperationReceivedFromMaster.set(currentlyApplyingOperationReceivedFromMaster);
     }
+
+    @Override
+    public void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service) {
+        this.unsentOperationForMasterQueue = service;
+    }
+
+    @Override
+    public <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
+            OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
+        if (unsentOperationForMasterQueue != null) {
+            unsentOperationForMasterQueue.retrySendingLater(operationWithResult, sender);
+        }
+    }
 }

java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationsToMasterSender.java

@@ -7,10 +7,30 @@ import java.io.InputStreamReader;
 import java.io.OutputStream;
 import java.net.HttpURLConnection;
 import java.net.URL;
+import java.util.logging.Logger;
 
+import com.sap.sse.common.WithID;
 import com.sap.sse.util.HttpUrlConnectionHelper;
 
-public interface OperationsToMasterSender<S, O extends OperationWithResult<S, ?>> extends Replicable<S, O>, UnsentOperationsForMasterQueue {
+public interface OperationsToMasterSender<S, O extends OperationWithResult<S, ?>> extends UnsentOperationsToMasterSender, WithID {
+    final Logger logger = Logger.getLogger(OperationsToMasterSender.class.getName());
+    
+    /**
+     * @return the descriptor of the master from which this replica is replicating this {@link Replicable}, or
+     *         {@code null} if this {@link Replicable} is currently running as a master.
+     */
+    ReplicationMasterDescriptor getMasterDescriptor();
+
+    /**
+     * Writes an operation to an output stream such that it can be read by {@link #readOperation}.
+     * 
+     * @param closeStream
+     *            if <code>true</code>, the stream will be closed after having written the operation; in any case, the
+     *            content written will be flushed to the <code>outputStream</code> so that the caller may continue to
+     *            invoke this method for other operations and/or on other replicables without producing corrupt data.
+     */
+    void writeOperation(OperationWithResult<?, ?> operation, OutputStream outputStream, boolean closeStream) throws IOException;
+
     /**
      * When a replica has initiated (not received through replication) an operation, this operation needs to be sent to
      * the master for execution from where it will replicate across the replication tree. This method uses the

java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/Replicable.java

@@ -8,8 +8,6 @@ import java.io.OutputStream;
 import java.net.MalformedURLException;
 import java.util.logging.Logger;
 
-
-import com.sap.sse.common.WithID;
 import com.sap.sse.operationaltransformation.Operation;
 import com.sap.sse.operationaltransformation.OperationWithTransformationSupport;
 import com.sap.sse.util.ObjectInputStreamResolvingAgainstCache;
@@ -57,7 +55,8 @@ import com.sap.sse.util.ThreadLocalTransporter;
  * @author Axel Uhl (D043530)
  *
  */
-public interface Replicable<S, O extends OperationWithResult<S, ?>> extends Replicator<S, O>, WithID {
+public interface Replicable<S, O extends OperationWithResult<S, ?>>
+extends OperationsToMasterSender<S, OperationWithResult<S, ?>>, Replicator<S, O> {
     static final Logger logger = Logger.getLogger(Replicable.class.getName());
     
     /**
@@ -102,12 +101,6 @@ public interface Replicable<S, O extends OperationWithResult<S, ?>> extends Repl
     
     void stoppedReplicatingFrom(ReplicationMasterDescriptor master);
 
-    /**
-     * @return the descriptor of the master from which this replica is replicating this {@link Replicable}, or
-     *         {@code null} if this {@link Replicable} is currently running as a master.
-     */
-    ReplicationMasterDescriptor getMasterDescriptor();
-
     /**
      * An operation execution listener must be able to process notifications of operations being executed that have
      * type <code>S</code> or any more specific type. The listener can achieve this also by accepting any type more general than
@@ -160,16 +153,6 @@ public interface Replicable<S, O extends OperationWithResult<S, ?>> extends Repl
      */
     O readOperation(InputStream inputStream) throws IOException, ClassNotFoundException;
 
-    /**
-     * Writes an operation to an output stream such that it can be read by {@link #readOperation}.
-     * 
-     * @param closeStream
-     *            if <code>true</code>, the stream will be closed after having written the operation; in any case, the
-     *            content written will be flushed to the <code>outputStream</code> so that the caller may continue to
-     *            invoke this method for other operations and/or on other replicables without producing corrupt data.
-     */
-    void writeOperation(OperationWithResult<?, ?> operation, OutputStream outputStream, boolean closeStream) throws IOException;
-
     /**
      * Checks if {@link #hasSentOperationToMaster(OperationWithResultWithIdWrapper) the operation was previously
      * sent to the master}. If so, the operation is ignored because it has been applied before to this replica.
@@ -265,4 +248,12 @@ public interface Replicable<S, O extends OperationWithResult<S, ?>> extends Repl
      */
     boolean hasSentOperationToMaster(OperationWithResult<S, ?> operation);
     
+    
+    /**
+     * Injects a service into this replicable that this instance of {@link UnsentOperationsToMasterSender} can use
+     * as a delegate to implement the {@link UnsentOperationsToMasterSender#retrySendingLater(OperationWithResult, OperationsToMasterSender)}
+     * method. This replicable may not be able to enqueue operations for re-trying sending to master after
+     * an error occurred unless this method has been used to announce the service.
+     */
+    void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service);
 }

java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/ReplicableWithObjectInputStream.java

@@ -13,7 +13,7 @@ import java.util.logging.Logger;
 import com.sap.sse.common.WithID;
 import com.sap.sse.operationaltransformation.Operation;
 
-public interface ReplicableWithObjectInputStream<S, O extends OperationWithResult<S, ?>> extends OperationsToMasterSender<S, O> {
+public interface ReplicableWithObjectInputStream<S, O extends OperationWithResult<S, ?>> extends Replicable<S, O> {
     static final Logger logger = Logger.getLogger(ReplicableWithObjectInputStream.class.getName());
     
     /**
@@ -123,7 +123,11 @@ public interface ReplicableWithObjectInputStream<S, O extends OperationWithResul
             try {
                 sendReplicaInitiatedOperationToMaster(castOperation);
             } catch (IOException e) {
-                logger.log(Level.SEVERE, "Exception trying to send operation "+operation+" to master for replication", e);
+                logger.log(Level.INFO, "Error sending operation "+operation+" to master "+getMasterDescriptor()+
+                        ". Queueing for later delivery.");
+                // remove the operation that failed to arrive on the master server from those marked as sent to master for now:
+                hasSentOperationToMaster(operation);
+                retrySendingLater(castOperation, this);
             }
         }
         replicateReplicated(operation); // this anticipates receiving the operation back from master; then, the operation will be ignored;

java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/UnsentOperationsToMasterSender.java

@@ -6,7 +6,7 @@ package com.sap.sse.replication;
  * @author Axel Uhl (d043530)
  *
  */
-public interface UnsentOperationsForMasterQueue {
+public interface UnsentOperationsToMasterSender {
     /**
      * When this replica was unable to send an operation to the master, e.g., for connectivity issues or the
      * master currently not being available, passing the operation to this method will enqueue the operation
@@ -16,5 +16,5 @@ public interface UnsentOperationsForMasterQueue {
      * 
      * @param sender the object to use to try to send the operation to the master server upon the next attempt
      */
-    <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(OperationWithResult<S, T> operationWithResultWithIdWrapper, OperationsToMasterSender<S, O> sender);
+    <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender);
 }

java/com.sap.sse.replication/src/com/sap/sse/replication/impl/ReplicationServiceImpl.java

@@ -43,7 +43,7 @@ import com.sap.sse.replication.ReplicablesProvider.ReplicableLifeCycleListener;
 import com.sap.sse.replication.ReplicationMasterDescriptor;
 import com.sap.sse.replication.ReplicationReceiver;
 import com.sap.sse.replication.ReplicationService;
-import com.sap.sse.replication.UnsentOperationsForMasterQueue;
+import com.sap.sse.replication.UnsentOperationsToMasterSender;
 import com.sap.sse.util.HttpUrlConnectionHelper;
 
 import net.jpountz.lz4.LZ4BlockInputStream;
@@ -76,7 +76,7 @@ import net.jpountz.lz4.LZ4BlockOutputStream;
  * @author Frank Mittag, Axel Uhl (d043530)
  * 
  */
-public class ReplicationServiceImpl implements ReplicationService, UnsentOperationsForMasterQueue {
+public class ReplicationServiceImpl implements ReplicationService, UnsentOperationsToMasterSender {
     private static final Logger logger = Logger.getLogger(ReplicationServiceImpl.class.getName());
 
     private final ReplicationInstancesManager replicationInstancesManager;
@@ -261,7 +261,7 @@ public class ReplicationServiceImpl implements ReplicationService, UnsentOperati
             synchronized (replicationInstancesManager) {
                 // .. and at least one of them wants to replicate the replicable with that ID
                 if (Util.contains(replicationInstancesManager.getAllReplicableIdsAtLeastOneReplicaIsReplicating(), replicable.getId().toString())) {
-                    ensureOperationExecutionListener(replicable);
+                    ensureOperationExecutionListenerAndInjectResetToMasterService(replicable);
                 }
             }
         }
@@ -383,7 +383,7 @@ public class ReplicationServiceImpl implements ReplicationService, UnsentOperati
     private void addAsListenerToReplicables(String[] replicableIdsAsStringForReplicablesToReplicate) {
         for (final String replicableIdAsStringForReplicableToReplicate : replicableIdsAsStringForReplicablesToReplicate) {
             Replicable<?, ?> replicable = getReplicable(replicableIdAsStringForReplicableToReplicate, /* wait */ true);
-            ensureOperationExecutionListener(replicable);
+            ensureOperationExecutionListenerAndInjectResetToMasterService(replicable);
         }
     }
 
@@ -392,11 +392,13 @@ public class ReplicationServiceImpl implements ReplicationService, UnsentOperati
      * {@code replicable} and remembered in {@link #executionListenersByReplicableIdAsString}. Otherwise, this method is
      * a no-op.
      */
-    private <S> void ensureOperationExecutionListener(Replicable<S, ?> replicable) {
+    private <S> void ensureOperationExecutionListenerAndInjectResetToMasterService(Replicable<S, ?> replicable) {
         if (!executionListenersByReplicableIdAsString.containsKey(replicable.getId().toString())) {
             final ReplicationServiceExecutionListener<S> listener = new ReplicationServiceExecutionListener<S>(this, replicable);
             executionListenersByReplicableIdAsString.put(replicable.getId().toString(), listener);
         }
+        // the following is idempotent and needs no roll-back
+        replicable.setUnsentOperationToMasterSender(unsentOperationsSenderJob);
     }
 
     @Override

java/com.sap.sse.replication/src/com/sap/sse/replication/impl/UnsentOperationsSenderJob.java

@@ -11,7 +11,7 @@ import com.sap.sse.common.Duration;
 import com.sap.sse.common.Util.Pair;
 import com.sap.sse.replication.OperationWithResult;
 import com.sap.sse.replication.OperationsToMasterSender;
-import com.sap.sse.replication.UnsentOperationsForMasterQueue;
+import com.sap.sse.replication.UnsentOperationsToMasterSender;
 import com.sap.sse.util.ThreadPoolUtil;
 
 /**
@@ -29,7 +29,7 @@ import com.sap.sse.util.ThreadPoolUtil;
  * @author Axel Uhl (d043530)
  *
  */
-public class UnsentOperationsSenderJob implements UnsentOperationsForMasterQueue, Runnable {
+public class UnsentOperationsSenderJob implements UnsentOperationsToMasterSender, Runnable {
     private static final Logger logger = Logger.getLogger(UnsentOperationsSenderJob.class.getName());
     private final static Duration MAX_WAIT_TIME_BETWEEN_ATTEMPTS = Duration.ONE_MINUTE;
     
@@ -65,6 +65,7 @@ public class UnsentOperationsSenderJob implements UnsentOperationsForMasterQueue
      */
     private synchronized void ensureScheduled() {
         if (!scheduled) {
+            logger.info("Scheduling replication operation re-send to master in "+nextWaitDuration);
             ThreadPoolUtil.INSTANCE.getDefaultBackgroundTaskThreadPoolExecutor().schedule(this,
                     /* delay */ nextWaitDuration.asMillis(), TimeUnit.MILLISECONDS);
             scheduled = true;

java/com.sap.sse.security/src/com/sap/sse/security/impl/SecurityServiceImpl.java

@@ -72,7 +72,10 @@ import com.sap.sse.mail.MailService;
 import com.sap.sse.replication.OperationExecutionListener;
 import com.sap.sse.replication.OperationWithResult;
 import com.sap.sse.replication.OperationWithResultWithIdWrapper;
+import com.sap.sse.replication.OperationsToMasterSender;
 import com.sap.sse.replication.ReplicationMasterDescriptor;
+import com.sap.sse.replication.ReplicationService;
+import com.sap.sse.replication.UnsentOperationsToMasterSender;
 import com.sap.sse.security.BearerAuthenticationToken;
 import com.sap.sse.security.ClientUtils;
 import com.sap.sse.security.Credential;
@@ -121,6 +124,14 @@ public class SecurityServiceImpl implements ReplicableSecurityService, ClearStat
     private ThreadLocal<Boolean> currentlyFillingFromInitialLoad = ThreadLocal.withInitial(() -> false);
     
     private ThreadLocal<Boolean> currentlyApplyingOperationReceivedFromMaster = ThreadLocal.withInitial(() -> false);
+    
+    /**
+     * This field is expected to be set by the {@link ReplicationService} once it has "adopted" this replicable.
+     * The {@link ReplicationService} "injects" this service so it can be used here as a delegate for the
+     * {@link UnsentOperationsToMasterSender#retrySendingLater(OperationWithResult, OperationsToMasterSender)}
+     * method.
+     */
+    private UnsentOperationsToMasterSender unsentOperationsToMasterSender;
 
     private static Ini shiroConfiguration;
     static {
@@ -1147,4 +1158,17 @@ public class SecurityServiceImpl implements ReplicableSecurityService, ClearStat
     public void removeAllSessions(String cacheName) {
         PersistenceFactory.INSTANCE.getDefaultMongoObjectFactory().removeAllSessions(cacheName);
     }
+
+    @Override
+    public void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service) {
+        this.unsentOperationsToMasterSender = service;
+    }
+
+    @Override
+    public <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
+            OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
+        if (unsentOperationsToMasterSender != null) {
+            unsentOperationsToMasterSender.retrySendingLater(operationWithResult, sender);
+        }
+    }
 }