eclipse-sailing-analytics/sailing-analytics
1
0
Fork 0
mirror of https://github.com/eclipse-sailing-analytics/sailing-analytics.git synced 2026-09-19 20:25:31 +00:00
Code Activity

ensure in-order delivery of messages from replica to master;

Browse Source 
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
This commit is contained in:
Axel Uhl
2019-01-22 14:31:58 +01:00
parent 112548b293
commit d2c3bdf2b5
15 changed files with 123 additions and 116 deletions
Download Patch File Download Diff File Expand all files Collapse all files
java/com.sap.sailing.domain.racelogtrackingadapter/src/com/sap/sailing/domain/racelogtracking/impl/fixtracker/RegattaLogFixTrackerRegattaListener.java
+6 -6
View File
@@ -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<RacingEventService, RacingEventService> 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 <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
public <S, O extends OperationWithResult<S, ?>, T> void scheduleForSending(
OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
if (unsentOperationsToMasterSender != null) {
unsentOperationsToMasterSender.retrySendingLater(operationWithResult, sender);
unsentOperationsToMasterSender.scheduleForSending(operationWithResult, sender);
}
}
}
java/com.sap.sailing.polars/src/com/sap/sailing/polars/impl/PolarDataServiceImpl.java
+6 -6
View File
@@ -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 <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
public <S, O extends OperationWithResult<S, ?>, T> void scheduleForSending(
OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
if (unsentOperationsToMasterSender != null) {
unsentOperationsToMasterSender.retrySendingLater(operationWithResult, sender);
unsentOperationsToMasterSender.scheduleForSending(operationWithResult, sender);
}
}
}
java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/UnsentOperationSenderJobTest.java
+13 -6
View File
@@ -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<RacingEventService, OperationWithResult<RacingEventService, String>> {
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<Ra
public void testResendThatFailsAtLeastOnce() throws InterruptedException, BrokenBarrierException {
final OperationWithResult<RacingEventService, String> operationWithResult = null;
final OperationsToMasterSender<RacingEventService, OperationWithResult<RacingEventService, String>> sender = this;
job.retrySendingLater(operationWithResult, sender);
job.scheduleForSending(operationWithResult, sender);
latch.await();
assertEquals(2, resendCount);
assertEquals(MAX_RESEND_COUNT+1, resendCount);
}
public <T> void sendReplicaInitiatedOperationToMaster(OperationWithResult<RacingEventService, T> 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<Ra
}
@Override
public <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
public <S, O extends OperationWithResult<S, ?>, T> void scheduleForSending(
OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
job.retrySendingLater(operationWithResult, sender);
job.scheduleForSending(operationWithResult, sender);
}
@Override
@@ -71,4 +75,7 @@ public class UnsentOperationSenderJobTest implements OperationsToMasterSender<Ra
@Override
public void addOperationSentToMasterForReplication(OperationWithResultWithIdWrapper<RacingEventService, ?> operationWithResultWithIdWrapper) {}
@Override
public boolean hasSentOperationToMaster(OperationWithResult<RacingEventService, ?> operation) { return false; }
}
java/com.sap.sailing.server/src/com/sap/sailing/server/impl/RacingEventServiceImpl.java
+6 -6
View File
@@ -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 <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
public <S, O extends OperationWithResult<S, ?>, T> void scheduleForSending(
OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
if (unsentOperationsToMasterSender != null) {
unsentOperationsToMasterSender.retrySendingLater(operationWithResult, sender);
unsentOperationsToMasterSender.scheduleForSending(operationWithResult, sender);
}
}
}
java/com.sap.sse.common/src/com/sap/sse/common/Duration.java
+1
View File
@@ -13,6 +13,7 @@ import com.sap.sse.common.impl.MillisecondsDurationImpl;
public interface Duration extends Serializable, Comparable<Duration> {
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);
java/com.sap.sse.filestorage/src/com/sap/sse/filestorage/impl/FileStorageManagementServiceImpl.java
+6 -6
View File
@@ -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<FileStorageService> 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 <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
public <S, O extends OperationWithResult<S, ?>, T> void scheduleForSending(
OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
if (unsentOperationForMasterQueue != null) {
unsentOperationForMasterQueue.retrySendingLater(operationWithResult, sender);
unsentOperationForMasterQueue.scheduleForSending(operationWithResult, sender);
}
}
}
java/com.sap.sse.mail/src/com/sap/sse/mail/impl/MailServiceImpl.java
+6 -6
View File
@@ -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 <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
public <S, O extends OperationWithResult<S, ?>, T> void scheduleForSending(
OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
if (unsentOperationForMasterQueue != null) {
unsentOperationForMasterQueue.retrySendingLater(operationWithResult, sender);
unsentOperationForMasterQueue.scheduleForSending(operationWithResult, sender);
}
}
}
java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationsToMasterSender.java
+8 -1
View File
@@ -11,7 +11,7 @@ import java.util.logging.Logger;
import com.sap.sse.common.WithID;
public interface OperationsToMasterSender<S, O extends OperationWithResult<S, ?>> extends UnsentOperationsToMasterSender, WithID {
public interface OperationsToMasterSender<S, O extends OperationWithResult<S, ?>> extends OperationsToMasterSendingQueue, WithID {
final Logger logger = Logger.getLogger(OperationsToMasterSender.class.getName());
/**
@@ -63,4 +63,11 @@ public interface OperationsToMasterSender<S, O extends OperationWithResult<S, ?>
* {@code operationWithResultWithIdWrapper} exactly once after this method has returned.
*/
void addOperationSentToMasterForReplication(OperationWithResultWithIdWrapper<S, ?> operationWithResultWithIdWrapper);
/**
* If an operation equal to <code>operationWithResultWithIdWrapper</code> has previously been passed to a call to
* {@link OperationsToMasterSender#addOperationSentToMasterForReplication(OperationWithResultWithIdWrapper)}, the call returns <code>true</code>
* exactly once.
*/
boolean hasSentOperationToMaster(OperationWithResult<S, ?> operation);
}
java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationsToMasterSendingQueue.java
Executable
+21
View File
@@ -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
*/
<S, O extends OperationWithResult<S, ?>, T> void scheduleForSending(OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender);
}
java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/Replicable.java
+3 -11
View File
@@ -242,18 +242,10 @@ extends OperationsToMasterSender<S, OperationWithResult<S, ?>>, Replicator<S, O>
}
/**
* If an operation equal to <code>operationWithResultWithIdWrapper</code> has previously been passed to a call to
* {@link OperationsToMasterSender#addOperationSentToMasterForReplication(OperationWithResultWithIdWrapper)}, the call returns <code>true</code>
* exactly once.
*/
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)}
* 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);
}
java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/ReplicableWithObjectInputStream.java
+7 -24
View File
@@ -120,15 +120,7 @@ public interface ReplicableWithObjectInputStream<S, O extends OperationWithResul
// if this is a replica and this replicable is not currently in the process of handling replication data (either an operation
// or the initial load) coming from the master, send the operation back to the master
if (getMasterDescriptor() != null && !isCurrentlyFillingFromInitialLoadOrApplyingOperationReceivedFromMaster()) {
try {
sendReplicaInitiatedOperationToMaster(castOperation);
} catch (IOException 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);
}
scheduleForSending(castOperation, this);
}
replicateReplicated(operation); // this anticipates receiving the operation back from master; then, the operation will be ignored;
// see also addOperationSentToMasterForReplication
@@ -173,7 +165,7 @@ public interface ReplicableWithObjectInputStream<S, O extends OperationWithResul
* Checks whether this replicable is a replica. If yes, the operation is executed locally and sent to the master
* server for execution. If sending the operation fails with an {@link IOException}, the operation will be enqueued
* for a later re-try using the
* {@link #retrySendingLater(OperationWithResultWithIdWrapper, OperationsToMasterSender)} method. Note that this may
* {@link #scheduleForSending(OperationWithResultWithIdWrapper, OperationsToMasterSender)} method. Note that this may
* also happen while in a resend attempt. Otherwise, {@link #applyReplicated(OperationWithResult)} is invoked which
* executes and replicates the operation immediately.
*/
@@ -185,20 +177,11 @@ public interface ReplicableWithObjectInputStream<S, O extends OperationWithResul
}
final T result = applyReplicated(operation);
ReplicationMasterDescriptor masterDescriptor = getMasterDescriptor();
try {
if (masterDescriptor != null) {
sendReplicaInitiatedOperationToMaster(operation);
}
} catch (IOException e) {
logger.log(Level.INFO, "Error sending operation "+operation+" to master "+masterDescriptor+
". 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(operation, this);
} finally {
if (needToRemoveThreadLocal) {
idOfOperationBeingExecuted.remove();
}
if (masterDescriptor != null) {
scheduleForSending(operation, this);
}
if (needToRemoveThreadLocal) {
idOfOperationBeingExecuted.remove();
}
return result;
}
java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/UnsentOperationsToMasterSender.java
-20
View File
@@ -1,20 +0,0 @@
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 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
* 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
*/
<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
+4 -4
View File
@@ -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 <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
unsentOperationsSenderJob.retrySendingLater(operationWithResult, sender);
public <S, O extends OperationWithResult<S, ?>, T> void scheduleForSending(OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
unsentOperationsSenderJob.scheduleForSending(operationWithResult, sender);
}
}
java/com.sap.sse.replication/src/com/sap/sse/replication/impl/UnsentOperationsSenderJob.java
+30 -14
View File
@@ -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 <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
public synchronized <S, O extends OperationWithResult<S, ?>, T> void scheduleForSending(
OperationWithResult<S, T> operationWithResult,
OperationsToMasterSender<S, O> 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<OperationWithResult<?, ?>, 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) {
java/com.sap.sse.security/src/com/sap/sse/security/impl/SecurityServiceImpl.java
+6 -6
View File
@@ -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 <S, O extends OperationWithResult<S, ?>, T> void retrySendingLater(
public <S, O extends OperationWithResult<S, ?>, T> void scheduleForSending(
OperationWithResult<S, T> operationWithResult, OperationsToMasterSender<S, O> sender) {
if (unsentOperationsToMasterSender != null) {
unsentOperationsToMasterSender.retrySendingLater(operationWithResult, sender);
unsentOperationsToMasterSender.scheduleForSending(operationWithResult, sender);
}
}
}