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

Merge branch 'bug4018'

Browse Source 
Change-Id: I68f5cb416a377aee3f8edb587f19f536add680e1
This commit is contained in:
Axel Uhl
2019-01-23 13:28:49 +01:00
parent a92aa70b87 2599fe22a8
commit 8c44936e2d
20 changed files with 337 additions and 160 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docker/Dockerfile.tpl
+3 -3
View File
@@ -2,15 +2,15 @@ FROM donaldduck70/sapjvm8:8.1.046
LABEL maintainer=axel.uhl@sap.com
# Download and extract the release
WORKDIR /home/sailing/servers/server
RUN wget -O /tmp/RELEASE.tar.gz http://releases.sapsailing.com/RELEASE/RELEASE.tar.gz \
&& tar xzvpf /tmp/RELEASE.tar.gz \
&& rm /tmp/RELEASE.tar.gz
RUN apt-get update \
&& apt-get install -y vim telnet dnsutils net-tools jq
COPY vimrc /root/.vimrc
RUN wget -O /tmp/rds.pem https://s3.amazonaws.com/rds-downloads/rds-combined-ca-bundle.pem \
&& /opt/sapjvm_8/bin/keytool -importcert -alias AWSRDS -file /tmp/rds.pem -keystore /opt/sapjvm_8/jre/lib/security/cacerts -noprompt -storepass changeit \
&& rm /tmp/rds.pem
RUN wget -O /tmp/RELEASE.tar.gz http://releases.sapsailing.com/RELEASE/RELEASE.tar.gz \
&& tar xzvpf /tmp/RELEASE.tar.gz \
&& rm /tmp/RELEASE.tar.gz
COPY env.sh .
COPY start .
COPY JavaSE-11.profile .
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/PostingOperationFromReplicaToMasterTest.java
Executable
+51
View File
@@ -0,0 +1,51 @@
package com.sap.sailing.server.replication.test;
import static org.junit.Assert.assertTrue;
import java.net.URISyntaxException;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.Timeout;
import com.sap.sailing.domain.base.CompetitorAndBoatStore;
import com.sap.sailing.domain.leaderboard.Leaderboard;
import com.sap.sailing.domain.leaderboard.impl.LowPoint;
import com.sap.sailing.server.operationaltransformation.AllowBoatResetToDefaults;
import com.sap.sailing.server.operationaltransformation.CreateFlexibleLeaderboard;
import com.sap.sailing.server.operationaltransformation.UpdateBoat;
/**
* Tests replication of boats in conjunction with the {@link CompetitorAndBoatStore} concepts, particularly the
* possibility to allow for boat data to be updated, either explicitly or implicitly from a tracking provider
* after marking the boat using
* {@link CompetitorAndBoatStore#allowBoatResetToDefaults(com.sap.sailing.domain.base.Boat)}.
*
* @author Frank Mittag
*
*/
public class PostingOperationFromReplicaToMasterTest extends AbstractServerReplicationTest {
@Rule public Timeout timeout = new Timeout(30, TimeUnit.SECONDS);
/**
* Add a tracked race to the master that includes a competitor; check that the boat was properly replicated to
* the replica's {@link CompetitorAndBoatStore}. Afterwards, use the {@link UpdateBoat} operation on the master to
* perform an explicit update; ensure that the update arrived on the replica. Then execute an
* {@link AllowBoatResetToDefaults} operation on the master, afterwards update the boat on the master,
* @throws URISyntaxException
*/
@Test
public void testPostOperationToMaster() throws InterruptedException, URISyntaxException {
final String leaderboardName = "My Leaderboard";
final CreateFlexibleLeaderboard operation = new CreateFlexibleLeaderboard(/* leaderboardName */ leaderboardName,
/* leaderboardDisplayName */ null, /* discardThresholds */ new int[0], /* scoringScheme */ new LowPoint(),
/* courseAreaId */ UUID.randomUUID());
replica.apply(operation);
Thread.sleep(1000);
assertTrue(master.getLeaderboards().containsKey(leaderboardName));
final Leaderboard masterLeaderboard = master.getLeaderboards().get(leaderboardName);
assertTrue(masterLeaderboard.getScoringScheme() instanceof LowPoint);
}
}
java/com.sap.sailing.server.replication.test/src/com/sap/sailing/server/replication/test/UnsentOperationSenderJobTest.java
+30 -7
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,31 +22,48 @@ 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() {
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 = new CountDownLatch(1);
job.scheduleForSending(operationWithResult, sender);
latch.await();
assertEquals(2, resendCount);
assertEquals(MAX_RESEND_COUNT+1, resendCount);
}
@Test
public void testSendingTwoOpsFollowedByAnotherOne() throws InterruptedException, BrokenBarrierException {
final OperationWithResult<RacingEventService, String> operationWithResult = null;
final OperationsToMasterSender<RacingEventService, OperationWithResult<RacingEventService, String>> sender = this;
latch = new CountDownLatch(2);
job.scheduleForSending(operationWithResult, sender);
job.scheduleForSending(operationWithResult, sender);
latch.await();
Thread.sleep(100);
assertEquals(MAX_RESEND_COUNT+2, resendCount);
latch = new CountDownLatch(1);
job.scheduleForSending(operationWithResult, sender);
latch.await();
assertEquals(MAX_RESEND_COUNT+3, 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 +71,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 +91,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.testsupport/META-INF/MANIFEST.MF
+7
View File
@@ -13,3 +13,10 @@ Require-Bundle: org.junit;bundle-version="4.8.2",
com.sap.sse.replication.interfaces;bundle-version="1.0.0"
Export-Package: com.sap.sse.replication.testsupport
Automatic-Module-Name: com.sap.sse.replication.testsupport
Import-Package: javax.servlet;version="3.1.0",
javax.servlet.http;version="3.1.0",
org.mockito;version="1.10.14",
org.mockito.invocation;version="1.10.14",
org.mockito.mock;version="1.10.14",
org.mockito.stubbing;version="1.10.14",
org.osgi.util.tracker;version="1.5.1"
java/com.sap.sse.replication.testsupport/src/com/sap/sse/replication/testsupport/AbstractServerReplicationTestSetUp.java
+110 -31
View File
@@ -2,9 +2,9 @@ package com.sap.sse.replication.testsupport;
import static org.junit.Assert.assertFalse;
import java.io.BufferedReader;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.io.OutputStreamWriter;
@@ -24,8 +24,18 @@ import java.util.UUID;
import java.util.logging.Level;
import java.util.logging.Logger;
import javax.servlet.ReadListener;
import javax.servlet.ServletException;
import javax.servlet.ServletInputStream;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import org.junit.Rule;
import org.junit.rules.Timeout;
import org.mockito.Matchers;
import org.mockito.Mockito;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;
@@ -42,6 +52,7 @@ import com.sap.sse.replication.impl.ReplicationInstancesManager;
import com.sap.sse.replication.impl.ReplicationMasterDescriptorImpl;
import com.sap.sse.replication.impl.ReplicationReceiverImpl;
import com.sap.sse.replication.impl.ReplicationServiceImpl;
import com.sap.sse.replication.impl.ReplicationServlet;
import com.sap.sse.replication.impl.SingletonReplicablesProvider;
import net.jpountz.lz4.LZ4BlockOutputStream;
@@ -276,44 +287,101 @@ public abstract class AbstractServerReplicationTestSetUp<ReplicableInterface ext
}
boolean stop = false;
while (!stop) {
Socket s = ss.accept();
String request = new BufferedReader(new InputStreamReader(s.getInputStream())).readLine();
final Socket s = ss.accept();
final InputStream inputStream = s.getInputStream();
String request = readLine(inputStream);
logger.info("received request "+request);
PrintWriter pw = new PrintWriter(new OutputStreamWriter(s.getOutputStream()));
pw.println("HTTP/1.1 200 OK");
pw.println("Content-Type: text/plain");
pw.println();
pw.flush();
if (request.contains("DEREGISTER")) {
// assuming that it is safe to unregister all replicas for tests
for (ReplicaDescriptor descriptor : getReplicaInfo()) {
unregisterReplica(descriptor);
if (request.startsWith("POST /replication/replication")) {
final ReplicationServlet servlet = new ReplicationServlet(new SingletonReplicablesProvider(master), /* replicationServiceTracker */ null);
final HttpServletRequest requestMock = Mockito.mock(HttpServletRequest.class);
Mockito.when(requestMock.getInputStream()).thenReturn(new ServletInputStream() {
@Override
public boolean isFinished() {
try {
return inputStream.available() <= 0;
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Override
public boolean isReady() {
try {
return inputStream.available() > 0;
} catch (IOException e) {
throw new RuntimeException(e);
}
}
@Override
public void setReadListener(ReadListener readListener) {
}
@Override
public int read() throws IOException {
return inputStream.read();
}
});
final HttpServletResponse responseMock = Mockito.mock(HttpServletResponse.class);
final boolean[] error = { false };
Mockito.doAnswer(new Answer<Void>() {
@Override
public Void answer(InvocationOnMock invocation) throws Throwable {
pw.println("HTTP/1.1 500 Bad Request");
pw.println("Content-Type: text/plain");
pw.println();
pw.println(invocation.getArgumentAt(1, String.class));
pw.flush();
error[0] = true;
return null;
}
}).when(responseMock).sendError(Matchers.anyInt(), Matchers.isA(String.class));
while (!readLine(inputStream).isEmpty());
servlet.doPost(requestMock, responseMock);
if (!error[0]) {
pw.println("HTTP/1.1 200 OK");
pw.println("Content-Type: text/plain");
pw.println();
pw.flush();
}
} else {
pw.println("HTTP/1.1 200 OK");
pw.println("Content-Type: text/plain");
pw.println();
pw.flush();
if (request.contains("DEREGISTER")) {
// assuming that it is safe to unregister all replicas for tests
for (ReplicaDescriptor descriptor : getReplicaInfo()) {
unregisterReplica(descriptor);
}
} else if (request.contains("REGISTER")) {
final String uuid = UUID.randomUUID().toString();
registerReplicaUuidForMaster(uuid, masterDescriptor);
masterReplicationService.registerReplica(replicaDescriptor);
pw.print(uuid);
} else if (request.contains("INITIAL_LOAD")) {
Channel channel = masterReplicationService.createMasterChannel();
RabbitOutputStream ros = new RabbitOutputStream(INITIAL_LOAD_PACKAGE_SIZE, channel,
/* queueName */ "initial-load-for-TestClient-"+UUID.randomUUID(), /* syncAfterTimeout */ false);
pw.println(ros.getQueueName());
final LZ4BlockOutputStream compressingOutputStream = new LZ4BlockOutputStream(ros);
master.serializeForInitialReplication(compressingOutputStream);
compressingOutputStream.finish();
ros.close();
} else if (request.contains("STOP")) {
stop = true;
logger.info("received STOP request");
}
} else if (request.contains("REGISTER")) {
final String uuid = UUID.randomUUID().toString();
registerReplicaUuidForMaster(uuid, masterDescriptor);
masterReplicationService.registerReplica(replicaDescriptor);
pw.print(uuid);
} else if (request.contains("INITIAL_LOAD")) {
Channel channel = masterReplicationService.createMasterChannel();
RabbitOutputStream ros = new RabbitOutputStream(INITIAL_LOAD_PACKAGE_SIZE, channel,
/* queueName */ "initial-load-for-TestClient-"+UUID.randomUUID(), /* syncAfterTimeout */ false);
pw.println(ros.getQueueName());
final LZ4BlockOutputStream compressingOutputStream = new LZ4BlockOutputStream(ros);
master.serializeForInitialReplication(compressingOutputStream);
compressingOutputStream.finish();
ros.close();
} else if (request.contains("STOP")) {
stop = true;
logger.info("received STOP request");
}
pw.close();
pw.flush(); // ensure response is delivered
s.close();
logger.info("Request handled successfully.");
}
} catch (IOException e) {
} catch (IOException | ServletException e) {
throw new RuntimeException(e);
} finally {
logger.info("replication servlet emulation done");
try {
if (ss != null) {
ss.close();
@@ -323,6 +391,17 @@ public abstract class AbstractServerReplicationTestSetUp<ReplicableInterface ext
}
}
}
private String readLine(final InputStream inputStream) throws IOException {
ByteArrayOutputStream bos = new ByteArrayOutputStream();
int b;
while ((b=inputStream.read()) != -1 && b != '\n') {
if (b != '\r') { // ignore CR
bos.write(b);
}
}
return new String(bos.toByteArray());
}
};
initialLoadTestServerThread.start();
synchronized (listening) {
java/com.sap.sse.replication/src/com/sap/sse/replication/impl/ReplicationServiceImpl.java
+11 -5
View File
@@ -9,6 +9,7 @@ import java.io.InputStreamReader;
import java.io.ObjectOutputStream;
import java.io.OutputStream;
import java.net.ConnectException;
import java.net.SocketException;
import java.net.URL;
import java.net.URLConnection;
import java.util.ArrayList;
@@ -36,6 +37,7 @@ import com.sap.sse.common.Util;
import com.sap.sse.replication.OperationExecutionListener;
import com.sap.sse.replication.OperationWithResult;
import com.sap.sse.replication.OperationsToMasterSender;
import com.sap.sse.replication.OperationsToMasterSendingQueue;
import com.sap.sse.replication.ReplicaDescriptor;
import com.sap.sse.replication.Replicable;
import com.sap.sse.replication.ReplicablesProvider;
@@ -43,7 +45,6 @@ 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.util.HttpUrlConnectionHelper;
import net.jpountz.lz4.LZ4BlockInputStream;
@@ -76,7 +77,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;
@@ -689,7 +690,12 @@ public class ReplicationServiceImpl implements ReplicationService, UnsentOperati
int read = content.read(buf);
while (read != -1) {
uuid.append(new String(buf, 0, read));
read = content.read(buf);
try {
read = content.read(buf);
} catch (SocketException e) {
// the connection may have been closed already; interpret this as the end of the stream
read = -1;
}
}
final String replicaUUID = uuid.toString();
registerReplicaUuidForMaster(replicaUUID, master);
@@ -831,7 +837,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/ReplicationServlet.java
+18 -8
View File
@@ -21,7 +21,6 @@ import javax.servlet.http.HttpServletResponse;
import org.apache.commons.lang.StringEscapeUtils;
import org.json.simple.JSONArray;
import org.json.simple.JSONObject;
import org.osgi.framework.BundleContext;
import org.osgi.util.tracker.ServiceTracker;
import com.rabbitmq.client.Channel;
@@ -29,6 +28,7 @@ import com.sap.sse.gateway.AbstractHttpServlet;
import com.sap.sse.replication.OperationWithResult;
import com.sap.sse.replication.ReplicaDescriptor;
import com.sap.sse.replication.Replicable;
import com.sap.sse.replication.ReplicablesProvider;
import com.sap.sse.replication.ReplicationReceiver;
import com.sap.sse.replication.ReplicationService;
import com.sap.sse.util.impl.CountingOutputStream;
@@ -72,17 +72,24 @@ public class ReplicationServlet extends AbstractHttpServlet {
private final ServiceTracker<ReplicationService, ReplicationService> replicationServiceTracker;
private final OSGiReplicableTracker replicablesProvider;
private final ReplicablesProvider replicablesProvider;
public ReplicationServlet() throws Exception {
BundleContext context = Activator.getDefaultContext();
replicablesProvider = new OSGiReplicableTracker(context);
replicationServiceTracker = new ServiceTracker<ReplicationService, ReplicationService>(context, ReplicationService.class.getName(), null);
replicationServiceTracker.open();
this(new OSGiReplicableTracker(Activator.getDefaultContext()),
new ServiceTracker<ReplicationService, ReplicationService>(Activator.getDefaultContext(),
ReplicationService.class.getName(), /* tracker customizer */ null));
}
public ReplicationServlet(ReplicablesProvider replicablesProvider, ServiceTracker<ReplicationService, ReplicationService> replicationServiceTracker) {
this.replicablesProvider = replicablesProvider;
this.replicationServiceTracker = replicationServiceTracker;
if (replicationServiceTracker != null) {
replicationServiceTracker.open();
}
}
protected ReplicationService getReplicationService() {
return replicationServiceTracker.getService();
return replicationServiceTracker == null ? null : replicationServiceTracker.getService();
}
/**
@@ -212,8 +219,11 @@ public class ReplicationServlet extends AbstractHttpServlet {
resp.getWriter().print(result.toJSONString());
}
/**
* Made public for test support
*/
@Override
protected void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
public void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
InputStream is = req.getInputStream();
DataInputStream dis = new DataInputStream(is);
String replicableIdAsString = dis.readUTF();
java/com.sap.sse.replication/src/com/sap/sse/replication/impl/UnsentOperationsSenderJob.java
+31 -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,30 @@ 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();
if (empty) {
scheduled = false;
}
}
}
}
@@ -104,8 +120,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 +131,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);
}
}
}