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 new file mode 100755 index 00000000000..78e243b69c6 --- /dev/null +++ b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationsToMasterSender.java @@ -0,0 +1,46 @@ +package com.sap.sse.replication; + +import java.io.BufferedReader; +import java.io.DataOutputStream; +import java.io.IOException; +import java.io.InputStreamReader; +import java.io.OutputStream; +import java.net.HttpURLConnection; +import java.net.URL; + +import com.sap.sse.util.HttpUrlConnectionHelper; + +public interface OperationsToMasterSender> extends Replicable, UnsentOperationsForMasterQueue { + /** + * When a replica has initiated (not received through replication) an operation, this operation needs to be sent to + * the master for execution from where it will replicate across the replication tree. This method uses the + * {@link ReplicationMasterDescriptor#getSendReplicaInitiatedOperationToMasterURL(String) URL for sending an + * operation to the replication servlet on the master} and through the POST request's output stream first sends the + * target replicable's ID as a string using a {@link DataOutputStream}, then + * {@link #writeOperation(OperationWithResult, OutputStream, boolean) serializes the operation}.

+ */ + default void sendReplicaInitiatedOperationToMaster(OperationWithResult operation) throws IOException { + ReplicationMasterDescriptor masterDescriptor = getMasterDescriptor(); + assert masterDescriptor != null; + final OperationWithResultWithIdWrapper operationWithResultWithIdWrapper = new OperationWithResultWithIdWrapper(operation); + // TODO bug4018: if sending the operation fails, e.g., because of an HTTP response code != 2xx, enqueue the operation for retry + addOperationSentToMasterForReplication(operationWithResultWithIdWrapper); + URL url = masterDescriptor.getSendReplicaInitiatedOperationToMasterURL(this.getId().toString()); + final HttpURLConnection connection = (HttpURLConnection) HttpUrlConnectionHelper.redirectConnection(url, "POST"); // sets doOutput to true + logger.info("Sending operation "+operation+" to master "+masterDescriptor+"'s replicable with ID "+this+" for initial execution and replication"); + connection.connect(); + OutputStream outputStream = connection.getOutputStream(); + DataOutputStream dos = new DataOutputStream(outputStream); + dos.writeUTF(getId().toString()); + this.writeOperation(operationWithResultWithIdWrapper, outputStream, /* closeStream */ true); + BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(connection.getInputStream())); + bufferedReader.close(); + } + + /** + * Records that the {@code operationWithResultWithIdWrapper} was sent to master for further processing and replication. + * The {@link #hasSentOperationToMaster(OperationWithResult)} will return {@code true} for an operation equal to + * {@code operationWithResultWithIdWrapper} exactly once after this method has returned. + */ + void addOperationSentToMasterForReplication(OperationWithResultWithIdWrapper operationWithResultWithIdWrapper); +} 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 21cd8dcb757..b9657d52600 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 @@ -260,7 +260,7 @@ public interface Replicable> extends Repl /** * If an operation equal to operationWithResultWithIdWrapper has previously been passed to a call to - * {@link #addOperationSentToMasterForReplication(OperationWithResultWithIdWrapper)}, the call returns true + * {@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/ReplicableWithObjectInputStream.java b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/ReplicableWithObjectInputStream.java index 0318c45a8b8..829d64393a0 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 @@ -1,25 +1,19 @@ package com.sap.sse.replication; -import java.io.BufferedReader; -import java.io.DataOutputStream; import java.io.IOException; import java.io.InputStream; -import java.io.InputStreamReader; import java.io.ObjectInputStream; import java.io.ObjectOutputStream; import java.io.OutputStream; import java.io.Serializable; -import java.net.HttpURLConnection; -import java.net.URL; import java.util.concurrent.atomic.AtomicInteger; import java.util.logging.Level; import java.util.logging.Logger; import com.sap.sse.common.WithID; import com.sap.sse.operationaltransformation.Operation; -import com.sap.sse.util.HttpUrlConnectionHelper; -public interface ReplicableWithObjectInputStream> extends Replicable { +public interface ReplicableWithObjectInputStream> extends OperationsToMasterSender { static final Logger logger = Logger.getLogger(ReplicableWithObjectInputStream.class.getName()); /** @@ -171,36 +165,13 @@ public interface ReplicableWithObjectInputStream> getOperationExecutionListeners(); - /** - * When a replica has initiated (not received through replication) an operation, this operation needs to be sent to - * the master for execution from where it will replicate across the replication tree. This method uses the - * {@link ReplicationMasterDescriptor#getSendReplicaInitiatedOperationToMasterURL(String) URL for sending an - * operation to the replication servlet on the master} and through the POST request's output stream first sends the - * target replicable's ID as a string using a {@link DataOutputStream}, then - * {@link #writeOperation(OperationWithResult, OutputStream, boolean) serializes the operation}. - */ - default void sendReplicaInitiatedOperationToMaster(OperationWithResult operation) throws IOException { - ReplicationMasterDescriptor masterDescriptor = getMasterDescriptor(); - assert masterDescriptor != null; - final OperationWithResultWithIdWrapper operationWithResultWithIdWrapper = new OperationWithResultWithIdWrapper(operation); - // TODO bug4018: if sending the operation fails, e.g., because of an HTTP response code != 2xx, enqueue the operation for retry - addOperationSentToMasterForReplication(operationWithResultWithIdWrapper); - URL url = masterDescriptor.getSendReplicaInitiatedOperationToMasterURL(this.getId().toString()); - final HttpURLConnection connection = (HttpURLConnection) HttpUrlConnectionHelper.redirectConnection(url, "POST"); // sets doOutput to true - logger.info("Sending operation "+operation+" to master "+masterDescriptor+"'s replicable with ID "+this+" for initial execution and replication"); - connection.connect(); - OutputStream outputStream = connection.getOutputStream(); - DataOutputStream dos = new DataOutputStream(outputStream); - dos.writeUTF(getId().toString()); - this.writeOperation(operationWithResultWithIdWrapper, outputStream, /* closeStream */ true); - BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(connection.getInputStream())); - bufferedReader.close(); - } - /** * Checks whether this replicable is a replica. If yes, the operation is executed locally and sent to the master - * server for execution. Otherwise, {@link #applyReplicated(OperationWithResult)} is invoked which executes and - * replicates the operation immediately. + * 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 + * also happen while in a resend attempt. Otherwise, {@link #applyReplicated(OperationWithResult)} is invoked which + * executes and replicates the operation immediately. */ default T apply(OperationWithResult operation) { boolean needToRemoveThreadLocal = false; @@ -214,8 +185,12 @@ public interface ReplicableWithObjectInputStream operationWithResultWithIdWrapper); - /** * The operation is executed by immediately {@link Operation#internalApplyTo(Object) applying} it to this * service object. It is then replicated to all replicas if and only if the operation is marked as diff --git a/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/UnsentOperationsForMasterQueue.java b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/UnsentOperationsForMasterQueue.java new file mode 100755 index 00000000000..f31a60e483e --- /dev/null +++ b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/UnsentOperationsForMasterQueue.java @@ -0,0 +1,20 @@ +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 UnsentOperationsForMasterQueue { + /** + * 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 retrySendingLater(OperationWithResult operationWithResultWithIdWrapper, 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 3ee0e8b4745..44c16ec311e 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 @@ -35,6 +35,7 @@ import com.sap.sse.ServerInfo; 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.ReplicaDescriptor; import com.sap.sse.replication.Replicable; import com.sap.sse.replication.ReplicablesProvider; @@ -42,6 +43,7 @@ import com.sap.sse.replication.ReplicablesProvider.ReplicableLifeCycleListener; import com.sap.sse.replication.ReplicationMasterDescriptor; import com.sap.sse.replication.ReplicationReceiver; import com.sap.sse.replication.ReplicationService; +import com.sap.sse.replication.UnsentOperationsForMasterQueue; import com.sap.sse.util.HttpUrlConnectionHelper; import net.jpountz.lz4.LZ4BlockInputStream; @@ -74,7 +76,7 @@ import net.jpountz.lz4.LZ4BlockOutputStream; * @author Frank Mittag, Axel Uhl (d043530) * */ -public class ReplicationServiceImpl implements ReplicationService { +public class ReplicationServiceImpl implements ReplicationService, UnsentOperationsForMasterQueue { private static final Logger logger = Logger.getLogger(ReplicationServiceImpl.class.getName()); private final ReplicationInstancesManager replicationInstancesManager; @@ -192,6 +194,8 @@ public class ReplicationServiceImpl implements ReplicationService { * */ private TimerTask sendingTask; + + private final UnsentOperationsSenderJob unsentOperationsSenderJob; /** * Defines for how many milliseconds the {@link #timer} will wait since the first operation has been added to an @@ -290,6 +294,7 @@ public class ReplicationServiceImpl implements ReplicationService { final ReplicationInstancesManager replicationInstancesManager, ReplicablesProvider replicablesProvider) throws IOException { timer = new Timer("ReplicationServiceImpl timer for delayed task sending", /* isDaemon */ true); + unsentOperationsSenderJob = new UnsentOperationsSenderJob(); executionListenersByReplicableIdAsString = new HashMap<>(); initialLoadChannels = new ConcurrentHashMap<>(); this.replicationInstancesManager = replicationInstancesManager; @@ -823,4 +828,9 @@ public class ReplicationServiceImpl implements ReplicationService { public boolean isReplicationStarting() { return this.replicationStarting; } + + @Override + public , T> void retrySendingLater(OperationWithResult operationWithResult, OperationsToMasterSender sender) { + unsentOperationsSenderJob.retrySendingLater(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 new file mode 100755 index 00000000000..0412f8bd155 --- /dev/null +++ b/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/UnsentOperationsSenderJob.java @@ -0,0 +1,124 @@ +package com.sap.sse.replication.impl; + +import java.io.IOException; +import java.util.Deque; +import java.util.concurrent.ConcurrentLinkedDeque; +import java.util.concurrent.TimeUnit; +import java.util.logging.Level; +import java.util.logging.Logger; + +import com.sap.sse.common.Duration; +import com.sap.sse.common.Util.Pair; +import com.sap.sse.replication.OperationWithResult; +import com.sap.sse.replication.OperationsToMasterSender; +import com.sap.sse.replication.UnsentOperationsForMasterQueue; +import com.sap.sse.util.ThreadPoolUtil; + +/** + * Manages a queue of operations that failed to get delivered to this replica's master server. + * When an operation is enqueued, it is added to the tail of the queue. Resend attempts start + * at the head of the queue, ensuring messages are delivered to the master in the order they + * were received by this job.

+ * + * If an operation is delivered successfully, delivering the next operation will be attempted + * immediately afterwards. If operation delivery fails, the job will re-schedule itself using the + * {@link ThreadPoolUtil#getDefaultBackgroundTaskThreadPoolExecutor() background thread pool executor} + * for a later time. The duration the job waits is increased gradually as more delivery attempts + * fail, up to a maximum wait time between attempts. + * + * @author Axel Uhl (d043530) + * + */ +public class UnsentOperationsSenderJob implements UnsentOperationsForMasterQueue, Runnable { + private static final Logger logger = Logger.getLogger(UnsentOperationsSenderJob.class.getName()); + private final static Duration MAX_WAIT_TIME_BETWEEN_ATTEMPTS = Duration.ONE_MINUTE; + + private final Deque, OperationsToMasterSender>> queue; + + private boolean scheduled; + + /** + * Something between {@link Duration#NULL} and {@link #MAX_WAIT_TIME_BETWEEN_ATTEMPTS}, indicating + * the time to wait before trying the next re-send. + */ + private Duration nextWaitDuration; + + public UnsentOperationsSenderJob() { + queue = new ConcurrentLinkedDeque<>(); + resetWaitDuration(); + } + + private void resetWaitDuration() { + nextWaitDuration = Duration.ONE_SECOND; + } + + @Override + public synchronized , T> void retrySendingLater( + OperationWithResult operationWithResult, + OperationsToMasterSender sender) { + queue.addLast(new Pair<>(operationWithResult, sender)); + ensureScheduled(); + } + + /** + * Ensures that this job is scheduled with an executor. + */ + private synchronized void ensureScheduled() { + if (!scheduled) { + ThreadPoolUtil.INSTANCE.getDefaultBackgroundTaskThreadPoolExecutor().schedule(this, + /* delay */ nextWaitDuration.asMillis(), TimeUnit.MILLISECONDS); + scheduled = true; + } + } + + @Override + public void run() { + synchronized (this) { // FIXME synchronization too coarse-grained + while (!queue.isEmpty() && tryToSend(queue.getFirst())) { + queue.removeFirst(); + resetWaitDuration(); + } + scheduled = false; + if (!queue.isEmpty()) { + ensureScheduled(); + } + } + } + + private , R> boolean tryToSend(Pair, OperationsToMasterSender> unsentOperationAndSender) { + @SuppressWarnings("unchecked") + OperationWithResult operation = (OperationWithResult) unsentOperationAndSender.getA(); + @SuppressWarnings("unchecked") + OperationsToMasterSender sender = (OperationsToMasterSender) unsentOperationAndSender.getB(); + return tryToSend(operation, sender); + } + + /** + * @return {@code true} if sending succeeded, {@code false} otherwise + */ + private , R> boolean tryToSend(OperationWithResult operation, OperationsToMasterSender sender) { + boolean result; + try { + sender.sendReplicaInitiatedOperationToMaster(operation); + result = true; + } catch (IOException e) { + result = false; + incrementWaitTime(); + logger.log(Level.INFO, "Error re-sending operation "+operation+" to master "+sender.getMasterDescriptor()+ + ". Will try again in "+nextWaitDuration); + } + return result; + } + + /** + * Doubles the wait duration, capping at {@link #MAX_WAIT_TIME_BETWEEN_ATTEMPTS}. + */ + private void incrementWaitTime() { + if (nextWaitDuration.compareTo(MAX_WAIT_TIME_BETWEEN_ATTEMPTS) < 0) { + nextWaitDuration = nextWaitDuration.times(2); + if (nextWaitDuration.compareTo(MAX_WAIT_TIME_BETWEEN_ATTEMPTS) > 0) { + nextWaitDuration = MAX_WAIT_TIME_BETWEEN_ATTEMPTS; + } + } + } +}