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

cleanly delete initial load queue when replication is stopped while initial load is running; catch Exception instead of IOException to also catch RuntimeExceptions

Browse Source
This commit is contained in:
Axel Uhl committed 2015-07-25 14:25:16 +02:00
1 parent 2cdbf55a41
commit 5a03c8392e
1 file changed
+16 -5
Download Patch File Download Diff File Expand all files Collapse all files
java/com.sap.sse.replication/src/com/sap/sse/replication/impl/ReplicationServiceImpl.java
+16 -5
View File
@@ -213,11 +213,13 @@ public class ReplicationServiceImpl implements ReplicationService {
private static class InitialLoadRequest {
private final Channel channelForInitialLoad;
private final Iterable<Replicable<?, ?>> replicables;
private final String queueName;
public InitialLoadRequest(Channel channelForInitialLoad, Iterable<Replicable<?, ?>> replicables) {
public InitialLoadRequest(Channel channelForInitialLoad, Iterable<Replicable<?, ?>> replicables, String queueName) {
super();
this.channelForInitialLoad = channelForInitialLoad;
this.replicables = replicables;
this.queueName = queueName;
}
public Channel getChannelForInitialLoad() {
@@ -227,6 +229,10 @@ public class ReplicationServiceImpl implements ReplicationService {
public Iterable<Replicable<?, ?>> getReplicables() {
return replicables;
}
protected String getQueueName() {
return queueName;
}
}
private class LifeCycleListener implements ReplicableLifeCycleListener {
@@ -574,7 +580,7 @@ public class ReplicationServiceImpl implements ReplicationService {
final String queueName = new BufferedReader(queueNameReader).readLine();
queueNameReader.close();
final Channel channel = master.createChannel();
initialLoadChannels.put(master, new InitialLoadRequest(channel, replicables));
initialLoadChannels.put(master, new InitialLoadRequest(channel, replicables, queueName));
final RabbitInputStreamProvider rabbitInputStreamProvider = new RabbitInputStreamProvider(channel, queueName);
try {
final GZIPInputStream gzipInputStream = new GZIPInputStream(rabbitInputStreamProvider.getInputStream());
@@ -685,9 +691,14 @@ public class ReplicationServiceImpl implements ReplicationService {
final InitialLoadRequest initialLoad = initialLoadChannels.get(descriptor);
if (initialLoad != null) {
try {
initialLoad.getChannelForInitialLoad().getConnection().close();
initialLoad.getChannelForInitialLoad().close();
} catch (IOException e) {
final Channel channelForInitialLoad = initialLoad.getChannelForInitialLoad();
// delete initial load queue
DeleteOk deleteOk = channelForInitialLoad.queueDelete(initialLoad.getQueueName());
logger.info("Deleted queue " + initialLoad.getQueueName()+ " used for initial load: " + deleteOk.toString());
initialLoadChannels.remove(descriptor);
channelForInitialLoad.getConnection().close();
channelForInitialLoad.close();
} catch (Exception 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