0003328: Detect when more than one instance of SymDS is connected the

same DB but not clustered.
JumpMind · Dec 4, 2017 · 1b97340 · 1b97340
1 parent 316b4f1
commit 1b97340
Show file tree

Hide file tree

Showing 13 changed files with 187 additions and 39 deletions.
diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/AbstractSymmetricEngine.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/AbstractSymmetricEngine.java
@@ -339,7 +339,8 @@ protected void init() {
         this.nodeService = new NodeService(this);
         this.configurationService = new ConfigurationService(parameterService, symmetricDialect,
                 nodeService);
-        this.clusterService = new ClusterService(parameterService, symmetricDialect);
+        this.dataService = new DataService(this, extensionService);
+        this.clusterService = new ClusterService(parameterService, symmetricDialect, nodeService);
         this.statisticService = new StatisticService(parameterService, symmetricDialect);
         this.statisticManager = createStatisticManager();
         this.concurrentConnectionManager = new ConcurrentConnectionManager(parameterService,
@@ -354,7 +355,6 @@ protected void init() {
         this.triggerRouterService = new TriggerRouterService(this);
         this.outgoingBatchService = new OutgoingBatchService(parameterService, symmetricDialect,
                 nodeService, configurationService, sequenceService, clusterService, extensionService);
-        this.dataService = new DataService(this, extensionService);
         this.routerService = buildRouterService();
         this.nodeCommunicationService = buildNodeCommunicationService(clusterService, nodeService, parameterService, configurationService, symmetricDialect);
         this.incomingBatchService = new IncomingBatchService(parameterService, symmetricDialect, clusterService);

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/common/ParameterConstants.java
@@ -283,6 +283,7 @@ private ParameterConstants() {
     public final static String CLUSTER_LOCK_REFRESH_MS = "cluster.lock.refresh.ms";
     public final static String LOCK_TIMEOUT_MS = "lock.timeout.ms";
     public final static String LOCK_WAIT_RETRY_MILLIS = "lock.wait.retry.ms";
+    public final static String INSTANCE_ID_LOCATION = "instance.id.location";
 
     public final static String PURGE_LOG_SUMMARY_MINUTES = "purge.log.summary.retention.minutes";
     public final static String PURGE_RETENTION_MINUTES = "purge.retention.minutes";

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/db/DatabaseUpgradeListener.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/db/DatabaseUpgradeListener.java
@@ -43,22 +43,17 @@ public class DatabaseUpgradeListener implements IDatabaseUpgradeListener, ISymme
 
     protected ISymmetricEngine engine;
 
-    protected boolean isUpgradeTo38;
+    protected boolean isUpgradeFromPre38;
+    protected boolean isUpgradeFrom38;
 
     @Override
     public String beforeUpgrade(ISymmetricDialect symmetricDialect, String tablePrefix, Database currentModel,
             Database desiredModel) throws IOException {
         StringBuilder sb = new StringBuilder();
-        String monitorTableName = tablePrefix + "_" + TableConstants.SYM_MONITOR;
-        String nodeTableName = tablePrefix + "_" + TableConstants.SYM_NODE;
-        if (currentModel.findTable(nodeTableName) != null && 
-                currentModel.findTable(monitorTableName) == null && desiredModel.findTable(monitorTableName) != null) {
-            log.info("Detected upgrade to version 3.8");
-            isUpgradeTo38 = true;
-        } else {
-            isUpgradeTo38 = false;
-        }
-        if (isUpgradeTo38) {
+
+        isUpgradeFromPre38 = isUpgradeFromPre38(tablePrefix, currentModel, desiredModel);
+
+        if (isUpgradeFromPre38) {
             Table transformTable = currentModel.findTable(tablePrefix + "_" + TableConstants.SYM_TRANSFORM_TABLE);
             if (transformTable != null && transformTable.findColumn("update_action") != null) {
                 engine.getSqlTemplate().update("update " + tablePrefix + "_" + TableConstants.SYM_TRANSFORM_TABLE +
@@ -109,10 +104,17 @@ public String beforeUpgrade(ISymmetricDialect symmetricDialect, String tablePref
         return sb.toString();
     }
 
+    private boolean isUpgradeFrom38(String tablePrefix, Database currentModel, Database desiredModel) {
+
+
+
+        return false;
+    }
+
     @Override
     public String afterUpgrade(ISymmetricDialect symmetricDialect, String tablePrefix, Database model) throws IOException {
         StringBuilder sb = new StringBuilder();
-        if (isUpgradeTo38) {
+        if (isUpgradeFromPre38) {
             engine.getSqlTemplate().update("update " + tablePrefix + "_" + TableConstants.SYM_SEQUENCE +
                     " set cache_size = 10 where sequence_name = ?", Constants.SEQUENCE_OUTGOING_BATCH);
             engine.getSqlTemplate().update("update  " + tablePrefix + "_" + TableConstants.SYM_CHANNEL +
@@ -150,6 +152,19 @@ protected void dropTriggers(Database currentModel, String tableName, String colu
         }
     }
 
+    protected boolean isUpgradeFromPre38(String tablePrefix, Database currentModel,
+            Database desiredModel) {
+        String monitorTableName = tablePrefix + "_" + TableConstants.SYM_MONITOR;
+        String nodeTableName = tablePrefix + "_" + TableConstants.SYM_NODE;
+        if (currentModel.findTable(nodeTableName) != null && 
+                currentModel.findTable(monitorTableName) == null && desiredModel.findTable(monitorTableName) != null) {
+            log.info("Detected upgrade from pre-3.8 version.");
+            return true;
+        } else {
+            return false;
+        }        
+    }
+
     @Override
     public void setSymmetricEngine(ISymmetricEngine engine) {
         this.engine = engine;

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/model/NodeHost.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/model/NodeHost.java
@@ -42,6 +42,7 @@ public class NodeHost implements Serializable {
 
     private String nodeId;
     private String hostName;
+    private String instanceId;
     private String ipAddress;
     private String osUser;
     private String osName;
@@ -64,12 +65,14 @@ public NodeHost() {
         this.createTime = new Date();
     }
 
-    public NodeHost(String nodeId) {
+    public NodeHost(String nodeId, String instanceId) {
         this();
         this.nodeId = nodeId;        
+        this.instanceId = instanceId;
     }        
 
-    public void refresh(IDatabasePlatform platform) {
+    public void refresh(IDatabasePlatform platform, String instanceId) {
+        this.instanceId = instanceId;
         this.hostName = AppUtils.getHostName();
         setIpAddress(AppUtils.getIpAddress());
         this.osUser = System.getProperty("user.name");
@@ -105,6 +108,14 @@ public String getHostName() {
     public void setHostName(String hostName) {
         this.hostName = hostName;
     }
+
+    public String getInstanceId() {
+        return instanceId;
+    }
+
+    public void setInstanceId(String instanceId) {
+        this.instanceId = instanceId;
+    }
 
     public String getIpAddress() {
         return ipAddress;

diff --git a/symmetric-core/src/main/java/org/jumpmind/symmetric/service/IClusterService.java b/symmetric-core/src/main/java/org/jumpmind/symmetric/service/IClusterService.java
@@ -60,5 +60,7 @@ public interface IClusterService {
     public boolean isInfiniteLocked(String action);
 
     public void persistToTableForSnapshot();
+
+    public String getInstanceId();
 
 }