diff --git a/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationWithResult.java b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationWithResult.java index 15d8d2339e7..7d817904bdc 100755 --- a/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationWithResult.java +++ b/java/com.sap.sse.replication.interfaces/src/com/sap/sse/replication/OperationWithResult.java @@ -1,5 +1,6 @@ package com.sap.sse.replication; +import java.io.ObjectOutputStream; import java.io.Serializable; import com.sap.sse.operationaltransformation.Operation; @@ -8,7 +9,21 @@ import com.sap.sse.operationaltransformation.Operation; * An operational transformation {@link Operation} is expected to return the target state after applying the operation. * This operation type offers the possibility to let an operation return a result value from its * {@link #internalApplyTo} method while in the context of operational transformation the operation will still return - * the target state. + * the target state.

+ * + * Operations that are instance of a class implementing this interface inherit the serializability declared by this + * interface. Multiple operation instances may be serialized into the same {@link ObjectOutputStream} (see also + * bug 5741). Keep this in mind when choosing how to represent the operation's internals. For example, it is not a + * good idea to have the operation reference a mutable object with the intention to transport a new "to-be" state + * this way. If multiple such state changes end up in the same stream of operations, the mutable object will be + * serialized to the stream only once, in one (probably the first, thus oldest) state, and all subsequent operations + * referencing that same object will serialize only a handle referencing the object state already written. With this, + * you may lose state changes. Hence, make sure to either clone such objects before storing the clone in the operation, + * or---even better---use specialized operations describing the specific state change to apply. For example, instead + * of serializing an object with all its attributes, rather implement an operation per attribute that updates only + * that attribute's value. It may be more tedious, but it keeps your serialized operations small, is therefore more + * bandwidth-efficient in the replication architecture and less prone to accidentally dropping state changes by + * representing the same object in the stream only once. * * @param * type of state to which the operation can be applied 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 9acff43e060..4c4dfe200ca 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 @@ -165,6 +165,12 @@ extends OperationsToMasterSender, Replicator { */ void clearReplicaState() throws MalformedURLException, IOException, InterruptedException; + /** + * Produces an object input stream that can choose to resolve objects against a cache so that duplicate instances + * are avoided. + */ + ObjectInputStream createObjectInputStreamResolvingAgainstCache(InputStream is, Map> classLoaderCache) throws IOException; + /** * Dual, reading operation for {@link #serializeForInitialReplication(OutputStream)}. In other words, when this * operation returns, this service instance is in a state "equivalent" to that of the service instance that produced @@ -187,6 +193,30 @@ extends OperationsToMasterSender, Replicator { */ void serializeForInitialReplication(OutputStream os) throws IOException; + /** + * The class loader to use for de-serializing objects. By default, this object's class's class loader is used. + */ + default ClassLoader getDeserializationClassLoader() { + return getClass().getClassLoader(); + } + + /** + * Implementation of {@link #readOperation(InputStream, Map)}, using the {@link ObjectInputStream} created by + * {@link #createObjectInputStreamResolvingAgainstCache(InputStream, Map)}. Before actually reading an operation + * object, the current thread's context class loader is set to the {@link #getDeserializationClassLoader() class + * loader for de-serialization} and restored to its previous value in the {@code finally} clause. + */ + @SuppressWarnings("unchecked") + default O readOperationFromObjectInputStream(ObjectInputStream ois) throws ClassNotFoundException, IOException { + ClassLoader oldContextClassloader = Thread.currentThread().getContextClassLoader(); + Thread.currentThread().setContextClassLoader(getDeserializationClassLoader()); + try { + return (O) ois.readObject(); + } finally { + Thread.currentThread().setContextClassLoader(oldContextClassloader); + } + } + /** * From an input stream, reads an operation that can be {@link #apply(OperationWithResult) applied} to this object. * Separating reading and applying gives clients an opportunity to queue operations, e.g., in order to wait until 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 a8e5a932779..59c66588e2c 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 @@ -32,12 +32,6 @@ public interface ReplicableWithObjectInputStream> classLoaderCache) throws IOException; - /** * Implementation of {@link #initiallyFillFrom(InputStream)} which receives an {@link ObjectInputStream} instead of * an {@link InputStream}. The {@link ObjectInputStream} is expected to have been produced by @@ -50,15 +44,6 @@ public interface ReplicableWithObjectInputStreamos by an {@link ObjectOutputStream} and invokes * {@link #serializeForInitialReplicationInternal(ObjectOutputStream)}. @@ -97,13 +75,7 @@ public interface ReplicableWithObjectInputStream> classLoaderCache) throws IOException, ClassNotFoundException { - ClassLoader oldContextClassloader = Thread.currentThread().getContextClassLoader(); - Thread.currentThread().setContextClassLoader(getDeserializationClassLoader()); - try { - return readOperationInternal(createObjectInputStreamResolvingAgainstCache(inputStream, classLoaderCache)); - } finally { - Thread.currentThread().setContextClassLoader(oldContextClassloader); - } + return readOperationFromObjectInputStream(createObjectInputStreamResolvingAgainstCache(inputStream, classLoaderCache)); } @Override diff --git a/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/ReplicationReceiverImpl.java b/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/ReplicationReceiverImpl.java index fbf8d953e3a..36db6d779cd 100755 --- a/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/ReplicationReceiverImpl.java +++ b/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/ReplicationReceiverImpl.java @@ -213,14 +213,12 @@ public class ReplicationReceiverImpl implements ReplicationReceiver, Runnable { // TODO bug 2465: decide based on the master descriptor whether we want to process this message; if it's for a Replicable we're not replicating from that master, drop the message Replicable replicable = replicableProvider.getReplicable(replicableIdAsString, /* wait */ false); if (replicable != null) { - ObjectInputStream ois = new ObjectInputStream(uncompressingInputStream); // no special stream required; only reading a generic byte[] + final ObjectInputStream ois = replicable.createObjectInputStreamResolvingAgainstCache(uncompressingInputStream, /* class loader cache */ new HashMap<>()); int operationsInMessage = 0; - final Map> classLoaderCache = new HashMap<>(); try { while (true) { - byte[] serializedOperation = (byte[]) ois.readObject(); + readOperationAndApplyOrQueueIt(replicable, ois); if (Util.contains(master.getReplicables(), replicable)) { - readOperationAndApplyOrQueueIt(replicable, serializedOperation, classLoaderCache); operationCount++; operationsInMessage++; if (operationCount % 10000l == 0) { @@ -374,9 +372,11 @@ public class ReplicationReceiverImpl implements ReplicationReceiver, Runnable { } private > void readOperationAndApplyOrQueueIt(Replicable replicable, - byte[] serializedOperation, Map> classLoaderCache) throws ClassNotFoundException, IOException { - O operation = replicable.readOperation(new ByteArrayInputStream(serializedOperation), classLoaderCache); - applyOrQueue(operation, replicable); + ObjectInputStream ois) throws ClassNotFoundException, IOException { + O operation = replicable.readOperationFromObjectInputStream(ois); + if (Util.contains(master.getReplicables(), replicable)) { + applyOrQueue(operation, replicable); + } } /** 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 9bc17d51bcd..b64fda2d29a 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 @@ -561,9 +561,6 @@ public class ReplicationServiceImpl implements ReplicationService, OperationsToM Replicable replicable) throws IOException { // need to write the operations one by one, making sure the ObjectOutputStream always writes // identical objects again if required because they may have changed state in between - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - replicable.writeOperation(operation, bos, /* close stream */true); - final byte[] serializedOperation = bos.toByteArray(); synchronized (outboundBufferMonitor) { final String replicaIdAsString = replicable.getId().toString(); if (outboundBuffer != null && !Util.equalsWithNull(outboundBufferReplicableIdAsString, replicaIdAsString)) { @@ -578,7 +575,7 @@ public class ReplicationServiceImpl implements ReplicationService, OperationsToM outboundObjectBuffer = compressingObjectOutputStream; outboundBufferClasses = new ArrayList<>(); } - outboundObjectBuffer.writeObject(serializedOperation); + outboundObjectBuffer.writeObject(operation); outboundBufferClasses.add(operation.getClassForLogging()); if (outboundBuffer.size() > TRIGGER_MESSAGE_SIZE_IN_BYTES) { logger.info("Triggering replication because buffer holds " + outboundBuffer.size()