mirror of
https://github.com/eclipse-sailing-analytics/sailing-analytics.git
synced 2026-11-13 06:02:10 +00:00
added hashCode/equals to ReplicationMasterDescriptorImpl; maintain initial load handles per master to stop them when shutting down
This commit is contained in:
Axel Uhl committed
1 parent
15988ee84f
commit
2cdbf55a41
3 files changed
+324
-166
No files matched your search
+238
-166
@@ -17,6 +17,7 @@ import java.util.Map;
|
||||
import java.util.Timer;
|
||||
import java.util.TimerTask;
|
||||
import java.util.UUID;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.logging.Level;
|
||||
import java.util.logging.Logger;
|
||||
import java.util.zip.GZIPInputStream;
|
||||
@@ -48,10 +49,10 @@ import com.sap.sse.replication.ReplicationService;
|
||||
*
|
||||
* The observers are registered only when there are replicas registered. If the last replica is de-registered, the
|
||||
* service stops observing the {@link Replicable}. Operations received that require replication are sent to the
|
||||
* {@link Exchange} to which replica queues can bind, using a {@link ReplicationReceiver}. By prefixing each message with the
|
||||
* {@link Object#toString()} representation of the {@link Replicable}'s {@link Replicable#getId() ID} the receiver can
|
||||
* determine to which {@link Replicable} to forward the operation. As such, this service multiplexes the replication
|
||||
* channels for potentially many {@link Replicable}s living in this server instance.
|
||||
* {@link Exchange} to which replica queues can bind, using a {@link ReplicationReceiver}. By prefixing each message
|
||||
* with the {@link Object#toString()} representation of the {@link Replicable}'s {@link Replicable#getId() ID} the
|
||||
* receiver can determine to which {@link Replicable} to forward the operation. As such, this service multiplexes the
|
||||
* replication channels for potentially many {@link Replicable}s living in this server instance.
|
||||
* <p>
|
||||
*
|
||||
* The exchange name and connectivity information for the message queuing system are provided to this service during
|
||||
@@ -68,135 +69,166 @@ import com.sap.sse.replication.ReplicationService;
|
||||
*/
|
||||
public class ReplicationServiceImpl implements ReplicationService {
|
||||
private static final Logger logger = Logger.getLogger(ReplicationServiceImpl.class.getName());
|
||||
|
||||
|
||||
private final ReplicationInstancesManager replicationInstancesManager;
|
||||
|
||||
|
||||
private final ReplicablesProvider replicablesProvider;
|
||||
|
||||
|
||||
/**
|
||||
* <code>null</code>, if this instance is not currently replicating from some master; the master's descriptor otherwise
|
||||
* <code>null</code>, if this instance is not currently replicating from some master; the master's descriptor
|
||||
* otherwise
|
||||
*/
|
||||
private ReplicationMasterDescriptor replicatingFromMaster;
|
||||
|
||||
|
||||
/**
|
||||
* The UUIDs with which this replica is registered by the master identified by the corresponding key
|
||||
*/
|
||||
private final Map<ReplicationMasterDescriptor, String> replicaUUIDs;
|
||||
|
||||
|
||||
/**
|
||||
* Channel used by a master server to publish replication operations; <code>null</code> in servers that don't have replicas registered
|
||||
* Channel used by a master server to publish replication operations; <code>null</code> in servers that don't have
|
||||
* replicas registered
|
||||
*/
|
||||
private Channel masterChannel;
|
||||
|
||||
|
||||
/**
|
||||
* The name of the RabbitMQ exchange to which this replication service sends its replication operations in
|
||||
* serialized form. Clients need to know this name to be able to bind their queues to the exchange.
|
||||
*/
|
||||
private final String exchangeName;
|
||||
|
||||
|
||||
/**
|
||||
* The host on which the RabbitMQ server runs
|
||||
*/
|
||||
private final String exchangeHost;
|
||||
|
||||
|
||||
/**
|
||||
* The port on which to reach the RabbitMQ server, or 0 for the RabbitMQ default port
|
||||
*/
|
||||
private final int exchangePort;
|
||||
|
||||
|
||||
/**
|
||||
* UUID that identifies this server
|
||||
*/
|
||||
private final UUID serverUUID;
|
||||
|
||||
|
||||
/**
|
||||
* For this instance running as a replica, the replicator receives messages from the master's queue and applies them
|
||||
* to the local replica.
|
||||
*/
|
||||
private ReplicationReceiver replicator;
|
||||
|
||||
|
||||
private final Map<String, ReplicationServiceExecutionListener<?>> executionListenersByReplicableIdAsString;
|
||||
|
||||
|
||||
private Thread replicatorThread;
|
||||
|
||||
|
||||
/**
|
||||
* Used to synchronize write access and replacements of {@link #outboundBuffer}, {@link #outboundObjectBuffer} and
|
||||
* {@link #outboundBufferClasses} when the timer scoops up the messages to send.
|
||||
*/
|
||||
private final Object outboundBufferMonitor = "";
|
||||
|
||||
|
||||
/**
|
||||
* Sending operations as serialized Java objects using binary RabbitMQ messages comes at an overhead. To reduce the overhead,
|
||||
* several operations can be serialized into a single message. The actual serialization of the buffer happens after a short duration
|
||||
* has passed since the last sending, managed by a {@link Timer}. Writers need to synchronize on {@link #outboundBufferMonitor}
|
||||
* which protects all of {@link #outboundBuffer}, {@link #outboundObjectBuffer} and {@link #outboundBufferClasses} which
|
||||
* are replaced or cleared when the timer scoops up the currently buffered operations to send them out.
|
||||
* Sending operations as serialized Java objects using binary RabbitMQ messages comes at an overhead. To reduce the
|
||||
* overhead, several operations can be serialized into a single message. The actual serialization of the buffer
|
||||
* happens after a short duration has passed since the last sending, managed by a {@link Timer}. Writers need to
|
||||
* synchronize on {@link #outboundBufferMonitor} which protects all of {@link #outboundBuffer},
|
||||
* {@link #outboundObjectBuffer} and {@link #outboundBufferClasses} which are replaced or cleared when the timer
|
||||
* scoops up the currently buffered operations to send them out.
|
||||
*/
|
||||
private ByteArrayOutputStream outboundBuffer;
|
||||
|
||||
|
||||
/**
|
||||
* The {@link #outboundBuffer} contains a message with serialized operations originating from a single
|
||||
* {@link Replicable} whose {@link Replicable#getId() ID} is written as its string value to the beginning
|
||||
* of the stream using {@link DataOutputStream#writeUTF(String)}. When an operation of a {@link Replicable}
|
||||
* with a different ID is to be {@link #broadcastOperation(OperationWithResult, Replicable) broadcast}, the
|
||||
* existing {@link #outboundBuffer} needs to be transmitted and a new one is started for the {@link Replicable}
|
||||
* now wanting to replicate an operation. Access is synchronized, as for {@link #outboundBuffer} using the
|
||||
* {@link Replicable} whose {@link Replicable#getId() ID} is written as its string value to the beginning of the
|
||||
* stream using {@link DataOutputStream#writeUTF(String)}. When an operation of a {@link Replicable} with a
|
||||
* different ID is to be {@link #broadcastOperation(OperationWithResult, Replicable) broadcast}, the existing
|
||||
* {@link #outboundBuffer} needs to be transmitted and a new one is started for the {@link Replicable} now wanting
|
||||
* to replicate an operation. Access is synchronized, as for {@link #outboundBuffer} using the
|
||||
* {@link #outboundBufferMonitor}.
|
||||
*/
|
||||
private String outboundBufferReplicableIdAsString;
|
||||
|
||||
/**
|
||||
* An object output stream that writes to {@link #outboundBuffer}. Operations are serialized into this stream until the timer
|
||||
* acquires the {@link #outboundBufferMonitor}, closes the stream and transmits the contents of {@link #outboundBuffer} as a
|
||||
* RabbitMQ message. While still holding the monitor, the timer task creates a new {@link #outboundBuffer} and a new
|
||||
* {@link #outboundObjectBuffer} wrapping the {@link #outboundBuffer}.
|
||||
* An object output stream that writes to {@link #outboundBuffer}. Operations are serialized into this stream until
|
||||
* the timer acquires the {@link #outboundBufferMonitor}, closes the stream and transmits the contents of
|
||||
* {@link #outboundBuffer} as a RabbitMQ message. While still holding the monitor, the timer task creates a new
|
||||
* {@link #outboundBuffer} and a new {@link #outboundObjectBuffer} wrapping the {@link #outboundBuffer}.
|
||||
*/
|
||||
private ObjectOutputStream outboundObjectBuffer;
|
||||
|
||||
|
||||
/**
|
||||
* Remembers the classes of the operations serialized into {@link #outboundObjectBuffer}. The list of classes in this list
|
||||
* matches with the sequence of objects written to {@link #outboundObjectBuffer} as long as the {@link #outboundBufferMonitor}
|
||||
* is being held.
|
||||
* Remembers the classes of the operations serialized into {@link #outboundObjectBuffer}. The list of classes in
|
||||
* this list matches with the sequence of objects written to {@link #outboundObjectBuffer} as long as the
|
||||
* {@link #outboundBufferMonitor} is being held.
|
||||
*/
|
||||
private List<Class<?>> outboundBufferClasses;
|
||||
|
||||
|
||||
/**
|
||||
* Used to schedule the sending of all operations in {@link #outboundBuffer} using the {@link #sendingTask}.
|
||||
*/
|
||||
private final Timer timer;
|
||||
|
||||
|
||||
/**
|
||||
* Sends all operations in {@link #outboundBuffer}. When holding the monitor of {@link #outboundBuffer},
|
||||
* the following rules hold:
|
||||
* Sends all operations in {@link #outboundBuffer}. When holding the monitor of {@link #outboundBuffer}, the
|
||||
* following rules hold:
|
||||
*
|
||||
* <ul>
|
||||
* <li>if <code>null</code>, adding an operation to {@link #outboundBuffer} needs to create and assign a new timer that
|
||||
* schedules a sending task.
|
||||
* </li>
|
||||
* <li>if not <code>null</code>, an operation added to {@link #outboundBuffer} is guaranteed to be sent by the timer
|
||||
* </li>
|
||||
* <li>if <code>null</code>, adding an operation to {@link #outboundBuffer} needs to create and assign a new timer
|
||||
* that schedules a sending task.</li>
|
||||
* <li>if not <code>null</code>, an operation added to {@link #outboundBuffer} is guaranteed to be sent by the timer
|
||||
* </li>
|
||||
* </ul>
|
||||
*
|
||||
*/
|
||||
private TimerTask sendingTask;
|
||||
|
||||
|
||||
/**
|
||||
* Defines for how many milliseconds the {@link #timer} will wait since the first operation has been added to an empty
|
||||
* {@link #outboundBuffer} until it carries out the actual transmission task. The longer this duration, the more operations
|
||||
* are likely to be sent per message transmitted, reducing overhead but correspondingly increasing latency.
|
||||
* Defines for how many milliseconds the {@link #timer} will wait since the first operation has been added to an
|
||||
* empty {@link #outboundBuffer} until it carries out the actual transmission task. The longer this duration, the
|
||||
* more operations are likely to be sent per message transmitted, reducing overhead but correspondingly increasing
|
||||
* latency.
|
||||
*/
|
||||
private final long TRANSMISSION_DELAY_MILLIS = 100;
|
||||
|
||||
|
||||
/**
|
||||
* Defines at which message size in bytes the message will be sent regardless the {@link #TRANSMISSION_DELAY_MILLIS}.
|
||||
* Defines at which message size in bytes the message will be sent regardless the {@link #TRANSMISSION_DELAY_MILLIS}
|
||||
* .
|
||||
*/
|
||||
private final int TRIGGER_MESSAGE_SIZE_IN_BYTES = 1024*1024;
|
||||
private final int TRIGGER_MESSAGE_SIZE_IN_BYTES = 1024 * 1024;
|
||||
|
||||
/**
|
||||
* Counts the messages sent out by this replicator
|
||||
*/
|
||||
private long messageCount;
|
||||
|
||||
|
||||
/**
|
||||
* Before receiving the initial load from a master is started, a message queue channel is opened through which the
|
||||
* initial load data stream is received. These channels are stored here for two reasons. If a master appears as key
|
||||
* here, an initial load process receiving from that master is currently running, and no second concurrent attempt
|
||||
* to replicate from that master should be started.
|
||||
*/
|
||||
private final Map<ReplicationMasterDescriptor, InitialLoadRequest> initialLoadChannels;
|
||||
|
||||
private static class InitialLoadRequest {
|
||||
private final Channel channelForInitialLoad;
|
||||
private final Iterable<Replicable<?, ?>> replicables;
|
||||
|
||||
public InitialLoadRequest(Channel channelForInitialLoad, Iterable<Replicable<?, ?>> replicables) {
|
||||
super();
|
||||
this.channelForInitialLoad = channelForInitialLoad;
|
||||
this.replicables = replicables;
|
||||
}
|
||||
|
||||
public Channel getChannelForInitialLoad() {
|
||||
return channelForInitialLoad;
|
||||
}
|
||||
|
||||
public Iterable<Replicable<?, ?>> getReplicables() {
|
||||
return replicables;
|
||||
}
|
||||
}
|
||||
|
||||
private class LifeCycleListener implements ReplicableLifeCycleListener {
|
||||
@Override
|
||||
public void replicableAdded(Replicable<?, ?> replicable) {
|
||||
@@ -210,13 +242,14 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
|
||||
@Override
|
||||
public void replicableRemoved(String replicableIdAsString) {
|
||||
ReplicationServiceExecutionListener<?> listener = executionListenersByReplicableIdAsString.remove(replicableIdAsString);
|
||||
ReplicationServiceExecutionListener<?> listener = executionListenersByReplicableIdAsString
|
||||
.remove(replicableIdAsString);
|
||||
if (listener != null) {
|
||||
listener.unsubscribe();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @param exchangeName
|
||||
* name of the fan-out exchange to which replication operations will be sent
|
||||
@@ -231,11 +264,12 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
* replicated when they have been removed, or start observing new {@link Recpliable} objects that were
|
||||
* added.
|
||||
*/
|
||||
public ReplicationServiceImpl(String exchangeName, String exchangeHost,
|
||||
int exchangePort, final ReplicationInstancesManager replicationInstancesManager,
|
||||
ReplicablesProvider replicablesProvider) throws IOException {
|
||||
public ReplicationServiceImpl(String exchangeName, String exchangeHost, int exchangePort,
|
||||
final ReplicationInstancesManager replicationInstancesManager, ReplicablesProvider replicablesProvider)
|
||||
throws IOException {
|
||||
timer = new Timer("ReplicationServiceImpl timer for delayed task sending");
|
||||
executionListenersByReplicableIdAsString = new HashMap<>();
|
||||
initialLoadChannels = new ConcurrentHashMap<>();
|
||||
this.replicationInstancesManager = replicationInstancesManager;
|
||||
replicaUUIDs = new HashMap<ReplicationMasterDescriptor, String>();
|
||||
this.replicablesProvider = replicablesProvider;
|
||||
@@ -247,30 +281,30 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
serverUUID = UUID.randomUUID();
|
||||
logger.info("Setting " + serverUUID.toString() + " as unique replication identifier.");
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
protected void finalize() {
|
||||
logger.info("terminating timer "+timer);
|
||||
logger.info("terminating timer " + timer);
|
||||
timer.cancel();
|
||||
}
|
||||
|
||||
protected ServiceTracker<Replicable<?, ?>, Replicable<?, ?>> getReplicableTracker() {
|
||||
return new ServiceTracker<Replicable<?, ?>, Replicable<?, ?>>(
|
||||
Activator.getDefaultContext(), Replicable.class.getName(), null);
|
||||
return new ServiceTracker<Replicable<?, ?>, Replicable<?, ?>>(Activator.getDefaultContext(),
|
||||
Replicable.class.getName(), null);
|
||||
}
|
||||
|
||||
|
||||
protected ReplicablesProvider getReplicablesProvider() {
|
||||
return replicablesProvider;
|
||||
}
|
||||
|
||||
|
||||
protected ReplicationReceiver getReplicator() {
|
||||
return replicator;
|
||||
}
|
||||
|
||||
|
||||
private Channel createMasterChannelAndDeclareFanoutExchange() throws IOException {
|
||||
Channel result = createMasterChannel();
|
||||
result.exchangeDeclare(exchangeName, "fanout");
|
||||
logger.info("Created fanout exchange "+exchangeName+" successfully.");
|
||||
logger.info("Created fanout exchange " + exchangeName + " successfully.");
|
||||
return result;
|
||||
}
|
||||
|
||||
@@ -286,7 +320,8 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
result = connectionFactory.newConnection().createChannel();
|
||||
} catch (ConnectException ex) {
|
||||
// make sure to log something meaningful
|
||||
logger.severe("Could not connect to messaging queue on " + connectionFactory.getHost() + ":" + connectionFactory.getPort() + "/" + exchangeName);
|
||||
logger.severe("Could not connect to messaging queue on " + connectionFactory.getHost() + ":"
|
||||
+ connectionFactory.getPort() + "/" + exchangeName);
|
||||
throw ex;
|
||||
}
|
||||
logger.info("Connected to " + connectionFactory.getHost() + ":" + connectionFactory.getPort());
|
||||
@@ -312,7 +347,7 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
}
|
||||
logger.info("Registered replica " + replica);
|
||||
}
|
||||
|
||||
|
||||
private void addAsListenerToReplicables() {
|
||||
for (Replicable<?, ?> replicable : getReplicables()) {
|
||||
addNewOperationExecutionListener(replicable);
|
||||
@@ -320,7 +355,8 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
}
|
||||
|
||||
private <S> void addNewOperationExecutionListener(Replicable<S, ?> replicable) {
|
||||
final ReplicationServiceExecutionListener<S> listener = new ReplicationServiceExecutionListener<S>(this, replicable);
|
||||
final ReplicationServiceExecutionListener<S> listener = new ReplicationServiceExecutionListener<S>(this,
|
||||
replicable);
|
||||
executionListenersByReplicableIdAsString.put(replicable.getId().toString(), listener);
|
||||
}
|
||||
|
||||
@@ -353,20 +389,23 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
* scheduled, a {@link #timer} is created and scheduled to send in {@link #TRANSMISSION_DELAY_MILLIS} milliseconds.
|
||||
*
|
||||
* @param replicable
|
||||
* the replicable by which the operation was executed that now will be broadcast to all replicas; the {@link Replicable#getId() ID}
|
||||
* of this replica in its string form will be used to prefix the operation
|
||||
* the replicable by which the operation was executed that now will be broadcast to all replicas; the
|
||||
* {@link Replicable#getId() ID} of this replica in its string form will be used to prefix the operation
|
||||
*/
|
||||
<S, O extends OperationWithResult<S, ?>> void broadcastOperation(OperationWithResult<?, ?> operation, Replicable<S, O> replicable) throws IOException {
|
||||
<S, O extends OperationWithResult<S, ?>> void broadcastOperation(OperationWithResult<?, ?> operation,
|
||||
Replicable<S, O> 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);
|
||||
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)) {
|
||||
flushBufferToRabbitMQ(); // operation from a replicable different from that for which operations are buffered so far --> flush
|
||||
} // still holding the monitor, so no other broadcast request from a different replicable can step in between
|
||||
flushBufferToRabbitMQ(); // operation from a replicable different from that for which operations are
|
||||
// buffered so far --> flush
|
||||
} // still holding the monitor, so no other broadcast request from a different replicable can step in
|
||||
// between
|
||||
if (outboundBuffer == null) {
|
||||
outboundBuffer = new ByteArrayOutputStream();
|
||||
outboundBufferReplicableIdAsString = replicaIdAsString;
|
||||
@@ -379,7 +418,7 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
outboundBufferClasses.add(operation.getClass());
|
||||
if (outboundBuffer.size() > TRIGGER_MESSAGE_SIZE_IN_BYTES) {
|
||||
logger.info("Triggering replication because buffer holds " + outboundBuffer.size()
|
||||
+ " bytes which exceeds trigger size " + TRIGGER_MESSAGE_SIZE_IN_BYTES+" bytes");
|
||||
+ " bytes which exceeds trigger size " + TRIGGER_MESSAGE_SIZE_IN_BYTES + " bytes");
|
||||
flushBufferToRabbitMQ();
|
||||
} else {
|
||||
if (sendingTask == null) {
|
||||
@@ -399,14 +438,16 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
}
|
||||
}
|
||||
if (++messageCount % 10000l == 0) {
|
||||
logger.info("Handled "+messageCount+" messages for replication. Current outbound replication queue size: "+outboundBufferClasses.size());
|
||||
logger.info("Handled " + messageCount
|
||||
+ " messages for replication. Current outbound replication queue size: "
|
||||
+ outboundBufferClasses.size());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Obtains the monitor on {@link #outboundBufferMonitor}, copies the references to the buffers, nulls out
|
||||
* the buffers, then releases the monitor and broadcasts the buffer.
|
||||
* Obtains the monitor on {@link #outboundBufferMonitor}, copies the references to the buffers, nulls out the
|
||||
* buffers, then releases the monitor and broadcasts the buffer.
|
||||
*/
|
||||
private void flushBufferToRabbitMQ() {
|
||||
logger.fine("Trying to acquire monitor");
|
||||
@@ -415,7 +456,8 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
final boolean doSend;
|
||||
synchronized (outboundBufferMonitor) {
|
||||
if (outboundBuffer != null) {
|
||||
logger.fine("Preparing " + outboundBufferClasses.size() + " operations for sending to RabbitMQ exchange");
|
||||
logger.fine("Preparing " + outboundBufferClasses.size()
|
||||
+ " operations for sending to RabbitMQ exchange");
|
||||
try {
|
||||
outboundObjectBuffer.close();
|
||||
logger.fine("Sucessfully closed ObjectOutputStream");
|
||||
@@ -443,18 +485,22 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
broadcastOperations(bytesToSend, classesOfOperationsToSend);
|
||||
logger.fine("Successfully handed " + classesOfOperationsToSend.size() + " operations to broadcaster");
|
||||
} catch (Exception e) {
|
||||
logger.log(Level.SEVERE, "Error trying to replicate " + classesOfOperationsToSend.size() + " operations", e);
|
||||
logger.log(Level.SEVERE, "Error trying to replicate " + classesOfOperationsToSend.size()
|
||||
+ " operations", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void broadcastOperations(byte[] bytesToSend, List<Class<?>> classesOfOperationsToSend) throws IOException {
|
||||
logger.fine("broadcasting "+classesOfOperationsToSend.size()+" operations as "+bytesToSend.length+" bytes");
|
||||
logger.fine("broadcasting " + classesOfOperationsToSend.size() + " operations as " + bytesToSend.length
|
||||
+ " bytes");
|
||||
if (masterChannel != null) {
|
||||
logger.fine("buffer to broadcast has "+bytesToSend.length+" bytes ("+(bytesToSend.length/1024/1024)+"MB)");
|
||||
logger.fine("buffer to broadcast has " + bytesToSend.length + " bytes ("
|
||||
+ (bytesToSend.length / 1024 / 1024) + "MB)");
|
||||
long startTime = System.currentTimeMillis();
|
||||
masterChannel.basicPublish(exchangeName, /* routingKey */"", /* properties */null, bytesToSend);
|
||||
logger.fine("successfully published "+bytesToSend.length+" bytes, taking "+(System.currentTimeMillis()-startTime)+"ms");
|
||||
logger.fine("successfully published " + bytesToSend.length + " bytes, taking "
|
||||
+ (System.currentTimeMillis() - startTime) + "ms");
|
||||
replicationInstancesManager.log(classesOfOperationsToSend, bytesToSend.length);
|
||||
}
|
||||
}
|
||||
@@ -475,99 +521,113 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
* which implements the initial load sending process.
|
||||
*/
|
||||
@Override
|
||||
public void startToReplicateFrom(ReplicationMasterDescriptor master) throws IOException, ClassNotFoundException, InterruptedException {
|
||||
public void startToReplicateFrom(ReplicationMasterDescriptor master) throws IOException, ClassNotFoundException,
|
||||
InterruptedException {
|
||||
final Iterable<Replicable<?, ?>> replicables = getReplicables();
|
||||
startToReplicateFrom(master, replicables);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void startToReplicateFrom(ReplicationMasterDescriptor master, Iterable<Replicable<?, ?>> replicables)
|
||||
public void startToReplicateFrom(final ReplicationMasterDescriptor master, final Iterable<Replicable<?, ?>> replicables)
|
||||
throws IOException, ClassNotFoundException, InterruptedException {
|
||||
logger.info("Starting to replicate from " + master);
|
||||
try {
|
||||
registerReplicaWithMaster(master);
|
||||
} catch (Exception ex) {
|
||||
logger.log(Level.SEVERE, "ERROR", ex);
|
||||
throw ex;
|
||||
}
|
||||
replicatingFromMaster = master;
|
||||
logger.info("Registered replica with master.");
|
||||
QueueingConsumer consumer = null;
|
||||
// logging exception here because it will not propagate
|
||||
// thru the client with all details
|
||||
try {
|
||||
logger.info("Connecting to message queue " + master);
|
||||
consumer = master.getConsumer();
|
||||
} catch (Exception ex) {
|
||||
logger.log(Level.SEVERE, "ERROR", ex);
|
||||
replicatingFromMaster = null;
|
||||
throw ex;
|
||||
}
|
||||
logger.info("Connection to exchange successful.");
|
||||
URL initialLoadURL = master.getInitialLoadURL(replicables);
|
||||
logger.info("Initial load URL is "+initialLoadURL);
|
||||
// start receiving messages already now, but start in suspended mode
|
||||
replicator = new ReplicationReceiver(master, replicablesProvider, /* startSuspended */true, consumer);
|
||||
// clear Replicable state here, before starting to receive and de-serialized operations which builds up
|
||||
// new state, e.g., in competitor store
|
||||
for (Replicable<?, ?> r : replicables) {
|
||||
r.clearReplicaState();
|
||||
r.startedReplicatingFrom(master);
|
||||
}
|
||||
replicatorThread = new Thread(replicator, "Replicator receiving from " + master.getMessagingHostname() + "/"
|
||||
+ master.getExchangeName());
|
||||
replicatorThread.start();
|
||||
logger.info("Started replicator thread");
|
||||
InputStream is = initialLoadURL.openStream();
|
||||
final String queueName = new BufferedReader(new InputStreamReader(is)).readLine();
|
||||
final Channel channel = master.createChannel();
|
||||
RabbitInputStreamProvider rabbitInputStreamProvider = new RabbitInputStreamProvider(channel, queueName);
|
||||
try {
|
||||
final GZIPInputStream gzipInputStream = new GZIPInputStream(rabbitInputStreamProvider.getInputStream());
|
||||
for (Replicable<?, ?> replicable : replicables) { // absolutely make sure to use the same sequence of replicables as for URL (bug 3015)
|
||||
logger.info("Starting to receive initial load for "+replicable.getId());
|
||||
try {
|
||||
replicable.initiallyFillFrom(gzipInputStream);
|
||||
} catch (Exception e) {
|
||||
logger.log(Level.SEVERE, "Exception trying to reveice initial load for "+replicable.getId(), e);
|
||||
throw e;
|
||||
if (initialLoadChannels.containsKey(master)) {
|
||||
logger.warning("An initial load from "+master+" is already running, replicating the following replicables: "+
|
||||
initialLoadChannels.get(master).getReplicables()+". Not starting a second time.");
|
||||
} else {
|
||||
logger.info("Starting to replicate from " + master);
|
||||
try {
|
||||
registerReplicaWithMaster(master);
|
||||
} catch (Exception ex) {
|
||||
logger.log(Level.SEVERE, "ERROR", ex);
|
||||
throw ex;
|
||||
}
|
||||
replicatingFromMaster = master;
|
||||
logger.info("Registered replica with master.");
|
||||
QueueingConsumer consumer = null;
|
||||
// logging exception here because it will not propagate
|
||||
// thru the client with all details
|
||||
try {
|
||||
logger.info("Connecting to message queue " + master);
|
||||
consumer = master.getConsumer();
|
||||
} catch (Exception ex) {
|
||||
logger.log(Level.SEVERE, "ERROR", ex);
|
||||
replicatingFromMaster = null;
|
||||
throw ex;
|
||||
}
|
||||
logger.info("Connection to exchange successful.");
|
||||
final URL initialLoadURL = master.getInitialLoadURL(replicables);
|
||||
logger.info("Initial load URL is " + initialLoadURL);
|
||||
// start receiving messages already now, but start in suspended mode
|
||||
replicator = new ReplicationReceiver(master, replicablesProvider, /* startSuspended */true, consumer);
|
||||
// clear Replicable state here, before starting to receive and de-serialize operations which builds up
|
||||
// new state, e.g., in competitor store
|
||||
for (Replicable<?, ?> r : replicables) {
|
||||
r.clearReplicaState();
|
||||
r.startedReplicatingFrom(master);
|
||||
}
|
||||
replicatorThread = new Thread(replicator, "Replicator receiving from " + master.getMessagingHostname() + "/"
|
||||
+ master.getExchangeName());
|
||||
replicatorThread.start();
|
||||
logger.info("Started replicator thread");
|
||||
final InputStream is = initialLoadURL.openStream();
|
||||
final InputStreamReader queueNameReader = new InputStreamReader(is);
|
||||
final String queueName = new BufferedReader(queueNameReader).readLine();
|
||||
queueNameReader.close();
|
||||
final Channel channel = master.createChannel();
|
||||
initialLoadChannels.put(master, new InitialLoadRequest(channel, replicables));
|
||||
final RabbitInputStreamProvider rabbitInputStreamProvider = new RabbitInputStreamProvider(channel, queueName);
|
||||
try {
|
||||
final GZIPInputStream gzipInputStream = new GZIPInputStream(rabbitInputStreamProvider.getInputStream());
|
||||
for (Replicable<?, ?> replicable : replicables) { // absolutely make sure to use the same sequence of
|
||||
// replicables as for URL (bug 3015)
|
||||
logger.info("Starting to receive initial load for " + replicable.getId());
|
||||
try {
|
||||
replicable.initiallyFillFrom(gzipInputStream);
|
||||
} catch (Exception e) {
|
||||
logger.log(Level.SEVERE, "Exception trying to reveice initial load for " + replicable.getId(), e);
|
||||
throw e;
|
||||
}
|
||||
logger.info("Done receiving initial load for " + replicable.getId());
|
||||
}
|
||||
logger.info("Done receiving initial load for "+replicable.getId());
|
||||
} finally {
|
||||
logger.info("Closing channel " + channel + "'s connection " + channel.getConnection());
|
||||
channel.getConnection().close();
|
||||
logger.info("Resuming replicator to apply queues");
|
||||
replicator.setSuspended(false); // apply queued operations
|
||||
// delete initial load queue
|
||||
DeleteOk deleteOk = consumer.getChannel().queueDelete(queueName);
|
||||
logger.info("Deleted queue " + queueName + " used for initial load: " + deleteOk.toString());
|
||||
initialLoadChannels.remove(master);
|
||||
}
|
||||
} finally {
|
||||
logger.info("Closing channel "+channel+"'s connection "+channel.getConnection());
|
||||
channel.getConnection().close();
|
||||
logger.info("Resuming replicator to apply queues");
|
||||
replicator.setSuspended(false); // apply queued operations
|
||||
// delete initial load queue
|
||||
DeleteOk deleteOk = consumer.getChannel().queueDelete(queueName);
|
||||
logger.info("Deleted queue " + queueName + " used for initial load: " + deleteOk.toString());
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @return the UUID that the master generated for this client which is also entered into {@link #replicaUUIDs}
|
||||
*/
|
||||
private String registerReplicaWithMaster(ReplicationMasterDescriptor master) throws IOException, ClassNotFoundException {
|
||||
URL replicationRegistrationRequestURL = master.getReplicationRegistrationRequestURL(getServerIdentifier(), BuildVersion.getBuildVersion());
|
||||
private String registerReplicaWithMaster(ReplicationMasterDescriptor master) throws IOException,
|
||||
ClassNotFoundException {
|
||||
URL replicationRegistrationRequestURL = master.getReplicationRegistrationRequestURL(getServerIdentifier(),
|
||||
BuildVersion.getBuildVersion());
|
||||
final URLConnection registrationRequestConnection = replicationRegistrationRequestURL.openConnection();
|
||||
registrationRequestConnection.connect();
|
||||
InputStream content = (InputStream) registrationRequestConnection.getContent();
|
||||
StringBuilder uuid = new StringBuilder();
|
||||
byte[] buf = new byte[256];
|
||||
final InputStream content = (InputStream) registrationRequestConnection.getContent();
|
||||
final StringBuilder uuid = new StringBuilder();
|
||||
final byte[] buf = new byte[256];
|
||||
int read = content.read(buf);
|
||||
while (read != -1) {
|
||||
uuid.append(new String(buf, 0, read));
|
||||
read = content.read(buf);
|
||||
}
|
||||
String replicaUUID = uuid.toString();
|
||||
final String replicaUUID = uuid.toString();
|
||||
registerReplicaUuidForMaster(replicaUUID, master);
|
||||
return replicaUUID;
|
||||
}
|
||||
|
||||
|
||||
protected void deregisterReplicaWithMaster(ReplicationMasterDescriptor master) {
|
||||
try {
|
||||
URL replicationDeRegistrationRequestURL = master.getReplicationDeRegistrationRequestURL(getServerIdentifier());
|
||||
URL replicationDeRegistrationRequestURL = master
|
||||
.getReplicationDeRegistrationRequestURL(getServerIdentifier());
|
||||
final URLConnection deregistrationRequestConnection = replicationDeRegistrationRequestURL.openConnection();
|
||||
deregistrationRequestConnection.connect();
|
||||
StringBuilder uuid = new StringBuilder();
|
||||
@@ -597,39 +657,51 @@ public class ReplicationServiceImpl implements ReplicationService {
|
||||
public Map<Class<? extends OperationWithResult<?, ?>>, Integer> getStatistics(ReplicaDescriptor replicaDescriptor) {
|
||||
return replicationInstancesManager.getStatistics(replicaDescriptor);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public double getAverageNumberOfOperationsPerMessage(ReplicaDescriptor replicaDescriptor) {
|
||||
return replicationInstancesManager.getAverageNumberOfOperationsPerMessage(replicaDescriptor);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public long getNumberOfMessagesSent(ReplicaDescriptor replica) {
|
||||
return replicationInstancesManager.getNumberOfMessagesSent(replica);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public long getNumberOfBytesSent(ReplicaDescriptor replica) {
|
||||
return replicationInstancesManager.getNumberOfBytesSent(replica);
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@Override
|
||||
public double getAverageNumberOfBytesPerMessage(ReplicaDescriptor replica) {
|
||||
return replicationInstancesManager.getAverageNumberOfBytesPerMessage(replica);
|
||||
}
|
||||
|
||||
|
||||
@Override
|
||||
public void stopToReplicateFromMaster() throws IOException {
|
||||
ReplicationMasterDescriptor descriptor = getReplicatingFromMaster();
|
||||
if (descriptor != null) {
|
||||
synchronized(replicaUUIDs) {
|
||||
final InitialLoadRequest initialLoad = initialLoadChannels.get(descriptor);
|
||||
if (initialLoad != null) {
|
||||
try {
|
||||
initialLoad.getChannelForInitialLoad().getConnection().close();
|
||||
initialLoad.getChannelForInitialLoad().close();
|
||||
} catch (IOException e) {
|
||||
logger.log(Level.WARNING, "Error trying to close initial load channel / connection to message queueing system", e);
|
||||
// let's continue trying to de-register the replica from master; re-throwing the
|
||||
// exception wouldn't make much sense here; rather try to clean up as much as
|
||||
// possible
|
||||
}
|
||||
}
|
||||
synchronized (replicaUUIDs) {
|
||||
if (replicator != null) {
|
||||
replicator.stop();
|
||||
deregisterReplicaWithMaster(descriptor);
|
||||
replicatingFromMaster = null;
|
||||
replicaUUIDs.clear();
|
||||
|
||||
// this is needed because QueuingConsumer.nextDelivery() wont unblock
|
||||
|
||||
// this is needed because QueuingConsumer.nextDelivery() won't unblock
|
||||
// if the connection is closed by application.
|
||||
replicatorThread.interrupt();
|
||||
replicator = null;
|
||||
|
||||
Reference in new issue
Block a user