From d2c3bdf2b5286e152689165afb305a2de888639b Mon Sep 17 00:00:00 2001 From: Axel Uhl Date: Tue, 22 Jan 2019 14:31:58 +0100 Subject: [PATCH] ensure in-order delivery of messages from replica to master; always pass operations to be sent to master into the queue so they cannot be delivered to the master earlier than other queued operations after connectivity has been restored Change-Id: I7b1409dc4663659039058ce2c0511a26009114b4 --- .../RegattaLogFixTrackerRegattaListener.java | 12 ++--- .../polars/impl/PolarDataServiceImpl.java | 12 ++--- .../test/UnsentOperationSenderJobTest.java | 19 +++++--- .../server/impl/RacingEventServiceImpl.java | 12 ++--- .../src/com/sap/sse/common/Duration.java | 1 + .../FileStorageManagementServiceImpl.java | 12 ++--- .../sap/sse/mail/impl/MailServiceImpl.java | 12 ++--- .../replication/OperationsToMasterSender.java | 9 +++- .../OperationsToMasterSendingQueue.java | 21 +++++++++ .../com/sap/sse/replication/Replicable.java | 14 ++---- .../ReplicableWithObjectInputStream.java | 31 +++---------- .../UnsentOperationsToMasterSender.java | 20 --------- .../impl/ReplicationServiceImpl.java | 8 ++-- .../impl/UnsentOperationsSenderJob.java | 44 +++++++++++++------ .../security/impl/SecurityServiceImpl.java | 12 ++--- 15 files changed, 123 insertions(+), 116 deletions(-) create mode 100755 java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationsToMasterSendingQueue.java delete mode 100755 java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/UnsentOperationsToMasterSender.java diff --git a/java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/RegattaLogFixTrackerRegattaListener.java b/java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/RegattaLogFixTrackerRegattaListener.java index 13dc6bb874b..200486cd9f5 100644 --- a/java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/RegattaLogFixTrackerRegattaListener.java +++ b/java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/RegattaLogFixTrackerRegattaListener.java @@ -31,7 +31,7 @@ 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; +import com.sap.sse.replication.OperationsToMasterSendingQueue; /** * This is the main entry point of the {@link SensorFixStore} based fix tracking. @@ -57,10 +57,10 @@ public class RegattaLogFixTrackerRegattaListener extends AbstractTrackedRegattaA /** * 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)} + * {@link OperationsToMasterSendingQueue#scheduleForSending(OperationWithResult, OperationsToMasterSender)} * method. */ - private UnsentOperationsToMasterSender unsentOperationsToMasterSender; + private OperationsToMasterSendingQueue unsentOperationsToMasterSender; public RegattaLogFixTrackerRegattaListener( ServiceTracker racingEventServiceTracker, @@ -228,15 +228,15 @@ public class RegattaLogFixTrackerRegattaListener extends AbstractTrackedRegattaA } @Override - public void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service) { + public void setUnsentOperationToMasterSender(OperationsToMasterSendingQueue service) { this.unsentOperationsToMasterSender = service; } @Override - public , T> void retrySendingLater( + public , T> void scheduleForSending( OperationWithResult operationWithResult, OperationsToMasterSender sender) { if (unsentOperationsToMasterSender != null) { - unsentOperationsToMasterSender.retrySendingLater(operationWithResult, sender); + unsentOperationsToMasterSender.scheduleForSending(operationWithResult, sender); } } } diff --git a/java/com.sap.sailing.polars/src/com/sap/sailing/polars/impl/PolarDataServiceImpl.java b/java/com.sap.sailing.polars/src/com/sap/sailing/polars/impl/PolarDataServiceImpl.java index bfb2fdbd997..d19966a2a7a 100755 --- a/java/com.sap.sailing.polars/src/com/sap/sailing/polars/impl/PolarDataServiceImpl.java +++ b/java/com.sap.sailing.polars/src/com/sap/sailing/polars/impl/PolarDataServiceImpl.java @@ -55,7 +55,7 @@ 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.replication.OperationsToMasterSendingQueue; import com.sap.sse.util.ClearStateTestSupport; /** @@ -92,10 +92,10 @@ public class PolarDataServiceImpl implements ReplicablePolarService, ClearStateT /** * 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)} + * {@link OperationsToMasterSendingQueue#scheduleForSending(OperationWithResult, OperationsToMasterSender)} * method. */ - private UnsentOperationsToMasterSender unsentOperationsToMasterSender; + private OperationsToMasterSendingQueue unsentOperationsToMasterSender; /** * Constructs the polar data service with default generation settings. @@ -434,15 +434,15 @@ public class PolarDataServiceImpl implements ReplicablePolarService, ClearStateT } @Override - public void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service) { + public void setUnsentOperationToMasterSender(OperationsToMasterSendingQueue service) { this.unsentOperationsToMasterSender = service; } @Override - public , T> void retrySendingLater( + public , T> void scheduleForSending( OperationWithResult operationWithResult, OperationsToMasterSender sender) { if (unsentOperationsToMasterSender != null) { - unsentOperationsToMasterSender.retrySendingLater(operationWithResult, sender); + unsentOperationsToMasterSender.scheduleForSending(operationWithResult, sender); } } } diff --git a/java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/UnsentOperationSenderJobTest.java b/java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/UnsentOperationSenderJobTest.java index 4237832c067..cf730c3e43c 100755 --- a/java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/UnsentOperationSenderJobTest.java +++ b/java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/UnsentOperationSenderJobTest.java @@ -7,9 +7,12 @@ import java.io.OutputStream; import java.io.Serializable; import java.util.concurrent.BrokenBarrierException; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; import org.junit.Before; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.Timeout; import com.sap.sailing.server.interfaces.RacingEventService; import com.sap.sse.replication.OperationWithResult; @@ -19,11 +22,12 @@ import com.sap.sse.replication.ReplicationMasterDescriptor; import com.sap.sse.replication.impl.UnsentOperationsSenderJob; public class UnsentOperationSenderJobTest implements OperationsToMasterSender> { + private static final int MAX_RESEND_COUNT = 10; private UnsentOperationsSenderJob job; private int resendCount; private CountDownLatch latch; -// @Rule public Timeout timeout = new Timeout(10, TimeUnit.SECONDS); + @Rule public Timeout timeout = new Timeout(10, TimeUnit.SECONDS); @Before public void setUp() { @@ -36,14 +40,14 @@ public class UnsentOperationSenderJobTest implements OperationsToMasterSender operationWithResult = null; final OperationsToMasterSender> sender = this; - job.retrySendingLater(operationWithResult, sender); + job.scheduleForSending(operationWithResult, sender); latch.await(); - assertEquals(2, resendCount); + assertEquals(MAX_RESEND_COUNT+1, resendCount); } public void sendReplicaInitiatedOperationToMaster(OperationWithResult operation) throws IOException { logger.info("resending at count "+resendCount); - if (resendCount++ == 0) { + if (resendCount++ < MAX_RESEND_COUNT) { throw new IOException("First resend failed"); } else { latch.countDown(); @@ -51,9 +55,9 @@ public class UnsentOperationSenderJobTest implements OperationsToMasterSender, T> void retrySendingLater( + public , T> void scheduleForSending( OperationWithResult operationWithResult, OperationsToMasterSender sender) { - job.retrySendingLater(operationWithResult, sender); + job.scheduleForSending(operationWithResult, sender); } @Override @@ -71,4 +75,7 @@ public class UnsentOperationSenderJobTest implements OperationsToMasterSender operationWithResultWithIdWrapper) {} + + @Override + public boolean hasSentOperationToMaster(OperationWithResult operation) { return false; } } diff --git a/java/com.sap.sailing.server/src/com/sap/sailing/server/impl/RacingEventServiceImpl.java b/java/com.sap.sailing.server/src/com/sap/sailing/server/impl/RacingEventServiceImpl.java index 81ae33b4f6e..abd55c3cd47 100755 --- a/java/com.sap.sailing.server/src/com/sap/sailing/server/impl/RacingEventServiceImpl.java +++ b/java/com.sap.sailing.server/src/com/sap/sailing/server/impl/RacingEventServiceImpl.java @@ -288,7 +288,7 @@ 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.replication.OperationsToMasterSendingQueue; import com.sap.sse.shared.media.ImageDescriptor; import com.sap.sse.shared.media.VideoDescriptor; import com.sap.sse.util.ClearStateTestSupport; @@ -516,10 +516,10 @@ public class RacingEventServiceImpl implements RacingEventService, ClearStateTes /** * 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)} + * {@link OperationsToMasterSendingQueue#scheduleForSending(OperationWithResult, OperationsToMasterSender)} * method. */ - private UnsentOperationsToMasterSender unsentOperationsToMasterSender; + private OperationsToMasterSendingQueue unsentOperationsToMasterSender; /** * Providing the constructor parameters for a new {@link RacingEventServiceImpl} instance is a bit tricky @@ -4365,15 +4365,15 @@ public class RacingEventServiceImpl implements RacingEventService, ClearStateTes } @Override - public void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service) { + public void setUnsentOperationToMasterSender(OperationsToMasterSendingQueue service) { this.unsentOperationsToMasterSender = service; } @Override - public , T> void retrySendingLater( + public , T> void scheduleForSending( OperationWithResult operationWithResult, OperationsToMasterSender sender) { if (unsentOperationsToMasterSender != null) { - unsentOperationsToMasterSender.retrySendingLater(operationWithResult, sender); + unsentOperationsToMasterSender.scheduleForSending(operationWithResult, sender); } } } diff --git a/java/com.sap.sse.common/src/com/sap/sse/common/Duration.java b/java/com.sap.sse.common/src/com/sap/sse/common/Duration.java index 78ed317937c..0df88394dce 100755 --- a/java/com.sap.sse.common/src/com/sap/sse/common/Duration.java +++ b/java/com.sap.sse.common/src/com/sap/sse/common/Duration.java @@ -13,6 +13,7 @@ import com.sap.sse.common.impl.MillisecondsDurationImpl; public interface Duration extends Serializable, Comparable { static final Duration NULL = new MillisecondsDurationImpl(0); + static final Duration ONE_MILLISECOND = new MillisecondsDurationImpl(1); static final Duration ONE_SECOND = new MillisecondsDurationImpl(1000); static final Duration ONE_MINUTE = ONE_SECOND.times(60); static final Duration ONE_HOUR = ONE_MINUTE.times(60); diff --git a/java/com.sap.sse.filestorage/src/com/sap/sse/filestorage/impl/FileStorageManagementServiceImpl.java b/java/com.sap.sse.filestorage/src/com/sap/sse/filestorage/impl/FileStorageManagementServiceImpl.java index 61b74bf6923..74ec3ff118e 100644 --- a/java/com.sap.sse.filestorage/src/com/sap/sse/filestorage/impl/FileStorageManagementServiceImpl.java +++ b/java/com.sap.sse.filestorage/src/com/sap/sse/filestorage/impl/FileStorageManagementServiceImpl.java @@ -29,7 +29,7 @@ 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.replication.OperationsToMasterSendingQueue; /** * Implements {@link ServiceTrackerCustomizer} so that all {@link FileStorageServices} announced in the registry can @@ -61,10 +61,10 @@ public class FileStorageManagementServiceImpl implements ReplicableFileStorageMa /** * 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)} + * {@link OperationsToMasterSendingQueue#scheduleForSending(OperationWithResult, OperationsToMasterSender)} * method. */ - private UnsentOperationsToMasterSender unsentOperationForMasterQueue; + private OperationsToMasterSendingQueue unsentOperationForMasterQueue; public FileStorageManagementServiceImpl(TypeBasedServiceFinder serviceFinder, FileStorageServicePropertyStore propertyStore) { @@ -248,15 +248,15 @@ public class FileStorageManagementServiceImpl implements ReplicableFileStorageMa } @Override - public void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service) { + public void setUnsentOperationToMasterSender(OperationsToMasterSendingQueue service) { this.unsentOperationForMasterQueue = service; } @Override - public , T> void retrySendingLater( + public , T> void scheduleForSending( OperationWithResult operationWithResult, OperationsToMasterSender sender) { if (unsentOperationForMasterQueue != null) { - unsentOperationForMasterQueue.retrySendingLater(operationWithResult, sender); + unsentOperationForMasterQueue.scheduleForSending(operationWithResult, sender); } } } diff --git a/java/com.sap.sse.mail/src/com/sap/sse/mail/impl/MailServiceImpl.java b/java/com.sap.sse.mail/src/com/sap/sse/mail/impl/MailServiceImpl.java index c5f19a385aa..3dce9ca661e 100644 --- a/java/com.sap.sse.mail/src/com/sap/sse/mail/impl/MailServiceImpl.java +++ b/java/com.sap.sse.mail/src/com/sap/sse/mail/impl/MailServiceImpl.java @@ -33,7 +33,7 @@ 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.replication.OperationsToMasterSendingQueue; import com.sap.sse.util.ObjectInputStreamResolvingAgainstCache; public class MailServiceImpl implements ReplicableMailService { @@ -57,10 +57,10 @@ public class MailServiceImpl implements ReplicableMailService { /** * 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)} + * {@link OperationsToMasterSendingQueue#scheduleForSending(OperationWithResult, OperationsToMasterSender)} * method. */ - private UnsentOperationsToMasterSender unsentOperationForMasterQueue; + private OperationsToMasterSendingQueue unsentOperationForMasterQueue; public MailServiceImpl(Properties mailProperties, MailServiceResolver mailServiceResolver) { this.mailProperties = mailProperties; @@ -256,15 +256,15 @@ public class MailServiceImpl implements ReplicableMailService { } @Override - public void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service) { + public void setUnsentOperationToMasterSender(OperationsToMasterSendingQueue service) { this.unsentOperationForMasterQueue = service; } @Override - public , T> void retrySendingLater( + public , T> void scheduleForSending( OperationWithResult operationWithResult, OperationsToMasterSender sender) { if (unsentOperationForMasterQueue != null) { - unsentOperationForMasterQueue.retrySendingLater(operationWithResult, sender); + unsentOperationForMasterQueue.scheduleForSending(operationWithResult, sender); } } } diff --git a/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationsToMasterSender.java b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationsToMasterSender.java index 3923ecdb5dd..058e42da17d 100755 --- a/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationsToMasterSender.java +++ b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationsToMasterSender.java @@ -11,7 +11,7 @@ import java.util.logging.Logger; import com.sap.sse.common.WithID; -public interface OperationsToMasterSender> extends UnsentOperationsToMasterSender, WithID { +public interface OperationsToMasterSender> extends OperationsToMasterSendingQueue, WithID { final Logger logger = Logger.getLogger(OperationsToMasterSender.class.getName()); /** @@ -63,4 +63,11 @@ public interface OperationsToMasterSender * {@code operationWithResultWithIdWrapper} exactly once after this method has returned. */ void addOperationSentToMasterForReplication(OperationWithResultWithIdWrapper operationWithResultWithIdWrapper); + + /** + * If an operation equal to operationWithResultWithIdWrapper has previously been passed to a call to + * {@link OperationsToMasterSender#addOperationSentToMasterForReplication(OperationWithResultWithIdWrapper)}, the call returns true + * exactly once. + */ + boolean hasSentOperationToMaster(OperationWithResult operation); } diff --git a/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationsToMasterSendingQueue.java b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationsToMasterSendingQueue.java new file mode 100755 index 00000000000..ba715d5d5c7 --- /dev/null +++ b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationsToMasterSendingQueue.java @@ -0,0 +1,21 @@ +package com.sap.sse.replication; + +/** + * Objects of this type can queue operations whose delivery to the master server has failed temporarily. + * + * @author Axel Uhl (d043530) + * + */ +public interface OperationsToMasterSendingQueue { + /** + * Adds the operation to the queue. A sending task is scheduled after the current send delay if none has been + * scheduled yet. 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 for + * later re-send attempts. Implementations have to make sure that operations are queued and sent in the order in + * which they are passed to this method. Therefore, implementations should {@code synchronize} on this object. + * + * @param sender + * the object to use to try to send the operation to the master server upon the next attempt + */ + , T> void scheduleForSending(OperationWithResult operationWithResult, OperationsToMasterSender sender); +} diff --git a/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/Replicable.java b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/Replicable.java index 6bb7ed4799e..b6260dcbef9 100755 --- a/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/Replicable.java +++ b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/Replicable.java @@ -242,18 +242,10 @@ extends OperationsToMasterSender>, Replicator } /** - * If an operation equal to operationWithResultWithIdWrapper has previously been passed to a call to - * {@link OperationsToMasterSender#addOperationSentToMasterForReplication(OperationWithResultWithIdWrapper)}, the call returns true - * exactly once. - */ - boolean hasSentOperationToMaster(OperationWithResult 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)} + * Injects a service into this replicable that this instance of {@link OperationsToMasterSendingQueue} can use + * as a delegate to implement the {@link OperationsToMasterSendingQueue#scheduleForSending(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); + void setUnsentOperationToMasterSender(OperationsToMasterSendingQueue service); } diff --git a/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/ReplicableWithObjectInputStream.java b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/ReplicableWithObjectInputStream.java index 2c9e3ee71b4..60a4732da28 100755 --- a/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/ReplicableWithObjectInputStream.java +++ b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/ReplicableWithObjectInputStream.java @@ -120,15 +120,7 @@ public interface ReplicableWithObjectInputStream, T> void retrySendingLater(OperationWithResult operationWithResult, OperationsToMasterSender sender); -} diff --git a/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/ReplicationServiceImpl.java b/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/ReplicationServiceImpl.java index 3a7d73e361f..715fada6863 100755 --- a/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/ReplicationServiceImpl.java +++ b/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.UnsentOperationsToMasterSender; +import com.sap.sse.replication.OperationsToMasterSendingQueue; 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, UnsentOperationsToMasterSender { +public class ReplicationServiceImpl implements ReplicationService, OperationsToMasterSendingQueue { private static final Logger logger = Logger.getLogger(ReplicationServiceImpl.class.getName()); private final ReplicationInstancesManager replicationInstancesManager; @@ -831,7 +831,7 @@ public class ReplicationServiceImpl implements ReplicationService, UnsentOperati } @Override - public , T> void retrySendingLater(OperationWithResult operationWithResult, OperationsToMasterSender sender) { - unsentOperationsSenderJob.retrySendingLater(operationWithResult, sender); + public , T> void scheduleForSending(OperationWithResult operationWithResult, OperationsToMasterSender sender) { + unsentOperationsSenderJob.scheduleForSending(operationWithResult, sender); } } diff --git a/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/UnsentOperationsSenderJob.java b/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/UnsentOperationsSenderJob.java index f88915947ce..0c5533ba2bd 100755 --- a/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/UnsentOperationsSenderJob.java +++ b/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.UnsentOperationsToMasterSender; +import com.sap.sse.replication.OperationsToMasterSendingQueue; import com.sap.sse.util.ThreadPoolUtil; /** @@ -29,7 +29,7 @@ import com.sap.sse.util.ThreadPoolUtil; * @author Axel Uhl (d043530) * */ -public class UnsentOperationsSenderJob implements UnsentOperationsToMasterSender, Runnable { +public class UnsentOperationsSenderJob implements OperationsToMasterSendingQueue, Runnable { private static final Logger logger = Logger.getLogger(UnsentOperationsSenderJob.class.getName()); private final static Duration MAX_WAIT_TIME_BETWEEN_ATTEMPTS = Duration.ONE_MINUTE; @@ -49,11 +49,11 @@ public class UnsentOperationsSenderJob implements UnsentOperationsToMasterSender } private void resetWaitDuration() { - nextWaitDuration = Duration.ONE_SECOND; + nextWaitDuration = Duration.ONE_MILLISECOND; } @Override - public synchronized , T> void retrySendingLater( + public synchronized , T> void scheduleForSending( OperationWithResult operationWithResult, OperationsToMasterSender sender) { queue.addLast(new Pair<>(operationWithResult, sender)); @@ -74,14 +74,29 @@ public class UnsentOperationsSenderJob implements UnsentOperationsToMasterSender @Override public void run() { - synchronized (this) { // FIXME synchronization too coarse-grained - while (!queue.isEmpty() && tryToSend(queue.getFirst())) { - queue.removeFirst(); - resetWaitDuration(); + boolean empty = false; + Pair, OperationsToMasterSender> first = null; + boolean sendOk = true; + while (!empty && sendOk) { + first = queue.peekFirst(); + if (first != null) { + sendOk = tryToSend(first); + } else { + sendOk = false; } - scheduled = false; - if (!queue.isEmpty()) { - ensureScheduled(); + synchronized (this) { // FIXME synchronization too coarse-grained + if (sendOk) { + resetWaitDuration(); + queue.removeFirst(); + } else { + incrementWaitDuration(); + // stop this loop (because sendOk==false) and re-schedule + scheduled = false; + ensureScheduled(); + } + empty = queue.isEmpty(); + assert empty || !sendOk; + assert empty || scheduled; } } } @@ -104,8 +119,9 @@ public class UnsentOperationsSenderJob implements UnsentOperationsToMasterSender result = true; } catch (IOException e) { result = false; - incrementWaitTime(); - logger.log(Level.INFO, "Error re-sending operation "+operation+" to master "+sender.getMasterDescriptor()+ + // remove the operation that failed to arrive on the master server from those marked as sent to master for now: + sender.hasSentOperationToMaster(operation); + logger.log(Level.INFO, "Error (re-)sending operation "+operation+" to master "+sender.getMasterDescriptor()+ ". Will try again in "+nextWaitDuration); } return result; @@ -114,7 +130,7 @@ public class UnsentOperationsSenderJob implements UnsentOperationsToMasterSender /** * Doubles the wait duration, capping at {@link #MAX_WAIT_TIME_BETWEEN_ATTEMPTS}. */ - private void incrementWaitTime() { + private void incrementWaitDuration() { if (nextWaitDuration.compareTo(MAX_WAIT_TIME_BETWEEN_ATTEMPTS) < 0) { nextWaitDuration = nextWaitDuration.times(2); if (nextWaitDuration.compareTo(MAX_WAIT_TIME_BETWEEN_ATTEMPTS) > 0) { diff --git a/java/com.sap.sse.security/src/com/sap/sse/security/impl/SecurityServiceImpl.java b/java/com.sap.sse.security/src/com/sap/sse/security/impl/SecurityServiceImpl.java index edd0821c0ac..4fb6af1e822 100644 --- a/java/com.sap.sse.security/src/com/sap/sse/security/impl/SecurityServiceImpl.java +++ b/java/com.sap.sse.security/src/com/sap/sse/security/impl/SecurityServiceImpl.java @@ -75,7 +75,7 @@ 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.replication.OperationsToMasterSendingQueue; import com.sap.sse.security.BearerAuthenticationToken; import com.sap.sse.security.ClientUtils; import com.sap.sse.security.Credential; @@ -128,10 +128,10 @@ public class SecurityServiceImpl implements ReplicableSecurityService, ClearStat /** * 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)} + * {@link OperationsToMasterSendingQueue#scheduleForSending(OperationWithResult, OperationsToMasterSender)} * method. */ - private UnsentOperationsToMasterSender unsentOperationsToMasterSender; + private OperationsToMasterSendingQueue unsentOperationsToMasterSender; private static Ini shiroConfiguration; static { @@ -1160,15 +1160,15 @@ public class SecurityServiceImpl implements ReplicableSecurityService, ClearStat } @Override - public void setUnsentOperationToMasterSender(UnsentOperationsToMasterSender service) { + public void setUnsentOperationToMasterSender(OperationsToMasterSendingQueue service) { this.unsentOperationsToMasterSender = service; } @Override - public , T> void retrySendingLater( + public , T> void scheduleForSending( OperationWithResult operationWithResult, OperationsToMasterSender sender) { if (unsentOperationsToMasterSender != null) { - unsentOperationsToMasterSender.retrySendingLater(operationWithResult, sender); + unsentOperationsToMasterSender.scheduleForSending(operationWithResult, sender); } } }