Provision MM at target MR

[dmaap/dbcapi.git] / src / main / java / org / onap / dmaap / dbcapi / service / MirrorMakerService.java

diff --git a/src/main/java/org/onap/dmaap/dbcapi/service/MirrorMakerService.java b/src/main/java/org/onap/dmaap/dbcapi/service/MirrorMakerService.java
index 8acc4f3..5d695f4 100644 (file)
--- a/src/main/java/org/onap/dmaap/dbcapi/service/MirrorMakerService.java
+++ b/src/main/java/org/onap/dmaap/dbcapi/service/MirrorMakerService.java
@@ -57,6 +57,7 @@ public class MirrorMakerService extends BaseLoggingClass {
        private static String defaultConsumerPort;
        private static String centralFqdn;
        private int maxTopicsPerMM;
+       private boolean mmPerMR;
        
        public MirrorMakerService() {
                super();
@@ -68,6 +69,7 @@ public class MirrorMakerService extends BaseLoggingClass {
                defaultConsumerPort = p.getProperty( "MR.TargetReplicationPort", "2181");       
                centralFqdn = p.getProperty("MR.CentralCname", "notSet");
                maxTopicsPerMM = Integer.valueOf( p.getProperty( "MaxTopicsPerMM", "5"));
+               mmPerMR = "true".equalsIgnoreCase(p.getProperty("MirrorMakerPerMR", "true"));
        }
 
        // will create a MM on MMagent if needed
@@ -79,34 +81,43 @@ public class MirrorMakerService extends BaseLoggingClass {
        
                DmaapService dmaap = new DmaapService();
                MR_ClusterService clusters = new MR_ClusterService();
-       
-               // in 1610, MM should only exist for edge-to-central
-               //  we use a cname for the central MR cluster that is active, and provision on agent topic on that target
-               // but only send 1 message so MM Agents can read it relying on kafka delivery
-               for( MR_Cluster central: clusters.getCentralClusters() ) {
-                       prov.makeTopicConnection(central, dmaap.getBridgeAdminFqtn(), centralFqdn  );
-                       ApiError resp = prov.doPostMessage(mm.createMirrorMaker( defaultConsumerPort, defaultProducerPort ));
-                       if ( ! resp.is2xx() ) {
-       
-                               errorLogger.error( DmaapbcLogMessageEnum.MM_PUBLISH_ERROR, "create MM", Integer.toString(resp.getCode()), resp.getMessage());
-                               mm.setStatus(DmaapObject_Status.INVALID);
-                       } else {
-                               prov.makeTopicConnection(central, dmaap.getBridgeAdminFqtn(), centralFqdn );
-                               resp = prov.doPostMessage(mm.getWhitelistUpdateJSON());
-                               if ( ! resp.is2xx()) {
-                                       errorLogger.error( DmaapbcLogMessageEnum.MM_PUBLISH_ERROR,"MR Bridge", Integer.toString(resp.getCode()), resp.getMessage());
-                                       mm.setStatus(DmaapObject_Status.INVALID);
-                               } else {
-                                       mm.setStatus(DmaapObject_Status.VALID);
-                               }
-                       }
-                       
-                       // we only want to send one message even if there are multiple central clusters
-                       break;
+               MR_Cluster target_cluster = null;
+               String override = null;
                
-               } 
+               if ( ! mmPerMR ) {
+                       // in ECOMP, MM Agent is only deployed at central, so this case is needed for backwards compatibility
+                       //  we use a cname for the central MR cluster that is active, and provision on agent topic on that target
+                       // but only send 1 message so MM Agents can read it relying on kafka delivery
+                       for( MR_Cluster cluster: clusters.getCentralClusters() ) {
+
+                               target_cluster = cluster;
+                               override = centralFqdn;
+                               // we only want to send one message even if there are multiple central clusters
+                               break;
+                       
+                       } 
+               } else {
+                       // In ONAP deployment architecture, the MM Agent is deployed with each target MR
+                       target_cluster = clusters.getMr_ClusterByFQDN(mm.getTargetCluster());
+                       override = null;
+               }
                
+               prov.makeTopicConnection(target_cluster, dmaap.getBridgeAdminFqtn(), override  );
+               ApiError resp = prov.doPostMessage(mm.createMirrorMaker( defaultConsumerPort, defaultProducerPort ));
+               if ( ! resp.is2xx() ) {
 
+                       errorLogger.error( DmaapbcLogMessageEnum.MM_PUBLISH_ERROR, "create MM", Integer.toString(resp.getCode()), resp.getMessage());
+                       mm.setStatus(DmaapObject_Status.INVALID);
+               } else {
+                       prov.makeTopicConnection(target_cluster, dmaap.getBridgeAdminFqtn(), override );
+                       resp = prov.doPostMessage(mm.getWhitelistUpdateJSON());
+                       if ( ! resp.is2xx()) {
+                               errorLogger.error( DmaapbcLogMessageEnum.MM_PUBLISH_ERROR,"MR Bridge", Integer.toString(resp.getCode()), resp.getMessage());
+                               mm.setStatus(DmaapObject_Status.INVALID);
+                       } else {
+                               mm.setStatus(DmaapObject_Status.VALID);
+                       }
+               }
 
                mm.setLastMod();
                return mirrors.put( mm.getMmName(), mm);
@@ -155,7 +166,7 @@ public class MirrorMakerService extends BaseLoggingClass {
                return ret;
        }
        
-       public MirrorMaker getNextMM( String source, String target ) {
+       public MirrorMaker getNextMM( String source, String target, String fqtn ) {
                int i = 0;
                MirrorMaker mm = null;
                while( mm == null ) {
@@ -164,6 +175,9 @@ public class MirrorMakerService extends BaseLoggingClass {
                        if ( mm == null ) {
                                mm = new MirrorMaker(source, target, i);
                        }
+                       if ( mm.getTopics().contains(fqtn) ) {
+                               break;
+                       }
                        if ( mm.getTopicCount() >= maxTopicsPerMM ) {
                                logger.info( "getNextMM: MM " + mm.getMmName() + " has " + mm.getTopicCount() + " topics.  Moving to next MM");
                                i++;
@@ -177,17 +191,17 @@ public class MirrorMakerService extends BaseLoggingClass {
 
        public MirrorMaker splitMM( MirrorMaker orig ) {
                
-               int index = 1;
                String source = orig.getSourceCluster();
                String target = orig.getTargetCluster();
                
                
                ArrayList<String> whitelist = orig.getTopics();
                while( whitelist.size() > maxTopicsPerMM ) {
-                       MirrorMaker mm = this.getNextMM( source, target );
+                       
                        int last = whitelist.size() - 1;
                        String topic = whitelist.get(last);
                        whitelist.remove(last);
+                       MirrorMaker mm = this.getNextMM( source, target, "aValueThatShouldNotMatchAnything" );
                        mm.addTopic(topic);     
                        this.updateMirrorMaker(mm);
                }