src/main/java/org/onap/dmaap/dbcapi/service/TopicService.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * org.onap.dmaap
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
   6  *
   7  * Modifications Copyright (C) 2019 IBM.
   8  * ================================================================================
   9  * Licensed under the Apache License, Version 2.0 (the "License");
  10  * you may not use this file except in compliance with the License.
  11  * You may obtain a copy of the License at
  12  *
  13  *      http://www.apache.org/licenses/LICENSE-2.0
  14  *
  15  * Unless required by applicable law or agreed to in writing, software
  16  * distributed under the License is distributed on an "AS IS" BASIS,
  17  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  18  * See the License for the specific language governing permissions and
  19  * limitations under the License.
  20  * ============LICENSE_END=========================================================
  21  */
  22
  23 package org.onap.dmaap.dbcapi.service;
  24
  25 import java.util.ArrayList;
  26 import java.util.Collection;
  27 import java.util.Iterator;
  28 import java.util.List;
  29 import java.util.Map;
  30 import java.util.Set;
  31
  32 import javax.ws.rs.core.Response.Status;
  33
  34 import org.onap.dmaap.dbcapi.aaf.AafNamespace;
  35 import org.onap.dmaap.dbcapi.aaf.AafRole;
  36 import org.onap.dmaap.dbcapi.aaf.AafService;
  37 import org.onap.dmaap.dbcapi.aaf.DmaapGrant;
  38 import org.onap.dmaap.dbcapi.aaf.AafService.ServiceType;
  39 import org.onap.dmaap.dbcapi.aaf.DmaapPerm;
  40 import org.onap.dmaap.dbcapi.database.DatabaseClass;
  41 import org.onap.dmaap.dbcapi.logging.BaseLoggingClass;
  42 import org.onap.dmaap.dbcapi.logging.DmaapbcLogMessageEnum;
  43 import org.onap.dmaap.dbcapi.model.ApiError;
  44 import org.onap.dmaap.dbcapi.model.DcaeLocation;
  45 import org.onap.dmaap.dbcapi.model.DmaapObject.DmaapObject_Status;
  46 import org.onap.dmaap.dbcapi.model.MR_Client;
  47 import org.onap.dmaap.dbcapi.model.MR_Cluster;
  48 import org.onap.dmaap.dbcapi.model.MirrorMaker;
  49 import org.onap.dmaap.dbcapi.model.ReplicationType;
  50 import org.onap.dmaap.dbcapi.model.Topic;
  51 import org.onap.dmaap.dbcapi.util.DmaapConfig;
  52 import org.onap.dmaap.dbcapi.util.Fqdn;
  53 import org.onap.dmaap.dbcapi.util.Graph;
  54
  55 public class TopicService extends BaseLoggingClass {
  56
  57
  58
  59         // REF: https://wiki.web.att.com/pages/viewpage.action?pageId=519703122
  60         private static String defaultGlobalMrHost;
  61
  62         private Map<String, Topic> mr_topics = DatabaseClass.getTopics();
  63
  64         private static DmaapService dmaapSvc = new DmaapService();
  65         private MR_ClientService clientService = new MR_ClientService();
  66         private MR_ClusterService clusters = new MR_ClusterService();
  67         private DcaeLocationService locations = new DcaeLocationService();
  68         private MirrorMakerService      bridge = new MirrorMakerService();
  69
  70         private static String centralCname;
  71         private static boolean createTopicRoles;
  72         private boolean strictGraph = true;
  73         private boolean mmPerMR;
  74
  75
  76         public TopicService(){
  77                 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
  78                 defaultGlobalMrHost = p.getProperty("MR.globalHost", "global.host.not.set");
  79                 centralCname = p.getProperty("MR.CentralCname");
  80                 createTopicRoles = "true".equalsIgnoreCase(p.getProperty("aaf.CreateTopicRoles", "true"));
  81                 String unit_test = p.getProperty( "UnitTest", "No" );
  82                 if ( unit_test.equals( "Yes" ) ) {
  83                         strictGraph = false;
  84                 }
  85                 mmPerMR = "true".equalsIgnoreCase(p.getProperty("MirrorMakerPerMR", "true"));
  86                 logger.info( "TopicService properties: CentralCname=" + centralCname +
  87                                 "   defaultGlobarlMrHost=" + defaultGlobalMrHost +
  88                                 " createTopicRoles=" + createTopicRoles +
  89                                 " mmPerMR=" + mmPerMR );
  90         }
  91
  92         public Map<String, Topic> getTopics() {
  93                 return mr_topics;
  94         }
  95
  96         public List<Topic> getAllTopics() {
  97                 return getAllTopics( true );
  98         }
  99         public List<Topic> getAllTopicsWithoutClients() {
 100                 return getAllTopics(false);
 101         }
 102
 103         private List<Topic> getAllTopics( Boolean withClients ) {
 104                 ArrayList<Topic> topics = new ArrayList<>(mr_topics.values());
 105                 if ( withClients ) {
 106                         for( Topic topic: topics ) {
 107                                 topic.setClients( clientService.getAllMrClients(topic.getFqtn()));
 108                         }
 109                 }
 110                 return topics;
 111         }
 112
 113
 114         public Topic getTopic( String key, ApiError apiError ) {
 115                 logger.info( "getTopic: key=" + key);
 116                 Topic t = mr_topics.get( key );
 117                 if ( t == null ) {
 118                         apiError.setCode(Status.NOT_FOUND.getStatusCode());
 119                         apiError.setFields( "fqtn");
 120                         apiError.setMessage("topic with fqtn " + key + " not found");
 121                         return null;
 122                 }
 123                 t.setClients( clientService.getAllMrClients( key ));
 124                 apiError.setCode(Status.OK.getStatusCode());
 125                 return t;
 126         }
 127
 128         private void aafTopicSetup(Topic topic, ApiError err ) {
 129
 130                 String nsr = dmaapSvc.getDmaap().getTopicNsRoot();
 131                 if ( nsr == null ) {
 132                         err.setCode(500);
 133                         err.setMessage("Unable to establish AAF namespace root: (check /dmaap object)"  );
 134                         err.setFields("topicNsRoot");
 135                         return;
 136                 }
 137
 138                 // establish AAF Connection using TopicMgr identity
 139                 AafService aaf = new AafService(ServiceType.AAF_TopicMgr);
 140
 141                 AafRole pubRole = null;
 142                 AafRole subRole = null;
 143
 144                 // creating Topic Roles was not an original feature.
 145                 // For backwards compatibility, only do this if the feature is enabled.
 146                 // Also, if the namespace of the topic is a foreign namespace, (i.e. not the same as our root ns)
 147                 // then we likely don't have permission to create sub-ns and Roles so don't try.
 148                 if ( createTopicRoles && topic.getFqtn().startsWith(nsr)) {
 149                         // create AAF namespace for this topic
 150                         AafNamespace ns = new AafNamespace( topic.getFqtn(), aaf.getIdentity());
 151                         {
 152                                 int rc = aaf.addNamespace( ns );
 153                                 if ( rc != 201 && rc != 409 ) {
 154                                         err.setCode(500);
 155                                         err.setMessage("Unexpected response from AAF:" + rc );
 156                                         err.setFields("namespace:" + topic.getFqtn() + " identity="+ aaf.getIdentity());
 157                                         return;
 158                                 }
 159                         }
 160
 161                         // create AAF Roles for MR clients of this topic
 162                         String rn = "publisher";
 163                         pubRole = new AafRole( topic.getFqtn(), rn );
 164                         int rc = aaf.addRole( pubRole );
 165                         if ( rc != 201 && rc != 409 ) {
 166                                 err.setCode(500);
 167                                 err.setMessage("Unexpected response from AAF:" + rc );
 168                                 err.setFields("topic:" + topic.getFqtn() + " role="+ rn);
 169                                 return;
 170                         }
 171                         topic.setPublisherRole( pubRole.getFullyQualifiedRole() );
 172
 173                         rn = "subscriber";
 174                         subRole = new AafRole( topic.getFqtn(), rn );
 175                         rc = aaf.addRole( subRole );
 176                         if ( rc != 201 && rc != 409 ) {
 177                                 err.setCode(500);
 178                                 err.setMessage("Unexpected response from AAF:" + rc );
 179                                 err.setFields("topic:" + topic.getFqtn() + " role="+ rn);
 180                                 return;
 181                         }
 182                         topic.setSubscriberRole( subRole.getFullyQualifiedRole() );
 183                 }
 184
 185                 // create AAF perms checked by MR
 186                 String instance = ":topic." + topic.getFqtn();
 187                 String[] actions = { "pub", "sub", "view" };
 188                 String t = dmaapSvc.getTopicPerm();
 189                 for ( String action : actions ){
 190                         DmaapPerm perm = new DmaapPerm( t, instance, action );
 191                         int rc = aaf.addPerm( perm );
 192                         if ( rc != 201 && rc != 409 ) {
 193                                 err.setCode(500);
 194                                 err.setMessage("Unexpected response from AAF:" + rc );
 195                                 err.setFields("t="+t + " instance="+ instance + " action="+ action);
 196                                 return;
 197                         }
 198                         if ( createTopicRoles ) {
 199                                 // Grant perms to our default Roles
 200                                 if ( action.equals( "pub") || action.equals( "view") ) {
 201                                         DmaapGrant g = new DmaapGrant( perm, pubRole.getFullyQualifiedRole() );
 202                                         rc = aaf.addGrant( g );
 203                                         if ( rc != 201 && rc != 409 ) {
 204                                                 err.setCode(rc);
 205                                                 err.setMessage( "Grant of " + perm.toString() + " failed for " + pubRole.getFullyQualifiedRole() );
 206                                                 logger.warn( err.getMessage());
 207                                                 return;
 208                                         }
 209                                 }
 210                                 if ( action.equals( "sub") || action.equals( "view") ) {
 211                                         DmaapGrant g = new DmaapGrant( perm, subRole.getFullyQualifiedRole() );
 212                                         rc = aaf.addGrant( g );
 213                                         if ( rc != 201 && rc != 409 ) {
 214                                                 err.setCode(rc);
 215                                                 err.setMessage( "Grant of " + perm.toString() + " failed for " + subRole.getFullyQualifiedRole() );
 216                                                 logger.warn( err.getMessage());
 217                                                 return;
 218                                         }
 219                                 }
 220                         }
 221
 222                 }
 223         }
 224
 225         public Topic addTopic( Topic topic, ApiError err, Boolean useExisting ) {
 226                 logger.info( "Entry: addTopic");
 227                 logger.info( "Topic name=" + topic.getTopicName() + " fqtnStyle=" + topic.getFqtnStyle() );
 228                 String nFqtn =  topic.genFqtn();
 229                 logger.info( "FQTN=" + nFqtn );
 230                 Topic pTopic = getTopic( nFqtn, err );
 231                 if ( pTopic != null ) {
 232                         String t = "topic already exists: " + nFqtn;
 233                         logger.info( t );
 234                         if (  useExisting ) {
 235                                 err.setCode(Status.OK.getStatusCode());
 236                                 return pTopic;
 237                         }
 238                         err.setMessage( t );
 239                         err.setFields( "fqtn");
 240                         err.setCode(Status.CONFLICT.getStatusCode());
 241                         return null;
 242                 }
 243                 err.reset();  // err filled with NOT_FOUND is expected case, but don't want to litter...
 244
 245                 topic.setFqtn( nFqtn );
 246
 247                 aafTopicSetup( topic, err );
 248                 if ( err.getCode() >= 400 ) {
 249                         return null;
 250                 }
 251
 252                 if ( topic.getReplicationCase().involvesGlobal() ) {
 253                         if ( topic.getGlobalMrURL() == null ) {
 254                                 topic.setGlobalMrURL(defaultGlobalMrHost);
 255                         }
 256                         if ( ! Fqdn.isValid( topic.getGlobalMrURL())) {
 257                                 logger.error( "GlobalMR FQDN not valid: " + topic.getGlobalMrURL());
 258                                 topic.setStatus( DmaapObject_Status.INVALID);
 259                                 err.setCode(500);
 260                                 err.setMessage("Value is not a valid FQDN:" +  topic.getGlobalMrURL() );
 261                                 err.setFields("globalMrURL");
 262
 263                                 return null;
 264                         }
 265                 }
 266
 267
 268                 if ( topic.getNumClients() > 0 ) {
 269                         ArrayList<MR_Client> clients = new ArrayList<MR_Client>(topic.getClients());
 270
 271
 272                         ArrayList<MR_Client> clients2 = new ArrayList<MR_Client>();
 273                         for ( Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
 274                                 MR_Client c = it.next();
 275
 276                                 logger.info( "c fqtn=" + c.getFqtn() + " ID=" + c.getMrClientId() + " url=" + c.getTopicURL());
 277                                 MR_Client nc = new MR_Client( c.getDcaeLocationName(), topic.getFqtn(), c.getClientRole(), c.getAction());
 278                                 nc.setFqtn(topic.getFqtn());
 279                                 nc.setClientIdentity( c.getClientIdentity());
 280                                 logger.info( "nc fqtn=" + nc.getFqtn() + " ID=" + nc.getMrClientId() + " url=" + nc.getTopicURL());
 281                                 clients2.add( clientService.addMr_Client(nc, topic, err));
 282                                 if ( ! err.is2xx()) {
 283                                         return null;
 284                                 }
 285                         }
 286
 287                         topic.setClients(clients2);
 288                 }
 289
 290                 Topic ntopic = checkForBridge( topic, err );
 291                 if ( ntopic == null ) {
 292                         topic.setStatus( DmaapObject_Status.INVALID);
 293                         if ( ! err.is2xx()) {
 294                                 return null;
 295                         }
 296                 }
 297
 298
 299                 mr_topics.put( nFqtn, ntopic );
 300
 301                 err.setCode(Status.OK.getStatusCode());
 302                 return ntopic;
 303         }
 304
 305
 306         public Topic updateTopic( Topic topic, ApiError err ) {
 307                 logger.info( "updateTopic: entry");
 308                 logger.info( "updateTopic: topic=" + topic);
 309                 logger.info( "updateTopic: fqtn=" + topic.getFqtn() );
 310                 if ( topic.getFqtn().isEmpty()) {
 311                         return null;
 312                 }
 313                 logger.info( "updateTopic: call checkForBridge");
 314                 Topic ntopic = checkForBridge( topic, err );
 315                 if ( ntopic == null ) {
 316                         topic.setStatus( DmaapObject_Status.INVALID);
 317                         if ( ! err.is2xx() ) {
 318                                 return null;
 319                         }
 320                 }
 321                 if(ntopic != null) {
 322                         logger.info( "updateTopic: call put");
 323                         mr_topics.put( ntopic.getFqtn(), ntopic );
 324                 }
 325                 err.setCode(Status.OK.getStatusCode());
 326                 return ntopic;
 327         }
 328
 329         public Topic removeTopic( String pubId, ApiError apiError ) {
 330                 Topic topic = mr_topics.get(pubId);
 331                 if ( topic == null ) {
 332                         apiError.setCode(Status.NOT_FOUND.getStatusCode());
 333                         apiError.setMessage("Topic " + pubId + " does not exist");
 334                         apiError.setFields("fqtn");
 335                         return null;
 336                 }
 337                 ArrayList<MR_Client> clients = new ArrayList<MR_Client>(clientService.getAllMrClients( pubId ));
 338                 for ( Iterator<MR_Client> it = clients.iterator(); it.hasNext(); ) {
 339                         MR_Client c = it.next();
 340
 341
 342                         clientService.removeMr_Client(c.getMrClientId(), false, apiError);
 343                         if ( ! apiError.is2xx()) {
 344                                 return null;
 345                         }
 346                 }
 347                 apiError.setCode(Status.OK.getStatusCode());
 348                 return mr_topics.remove(pubId);
 349         }
 350         public static ApiError setBridgeClientPerms( MR_Cluster node ) {
 351                 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
 352                 String mmProvRole = p.getProperty("MM.ProvRole");
 353                 String mmAgentRole = p.getProperty("MM.AgentRole");
 354                 String[] Roles = { mmProvRole, mmAgentRole };
 355                 String[] actions = { "view", "pub", "sub" };
 356                 Topic bridgeAdminTopic = new Topic().init();
 357                 bridgeAdminTopic.setTopicName( dmaapSvc.getBridgeAdminFqtn() );
 358                 bridgeAdminTopic.setTopicDescription( "RESERVED topic for MirroMaker Provisioning");
 359                 bridgeAdminTopic.setOwner( "DBC" );
 360
 361                 ArrayList<MR_Client> clients = new ArrayList<MR_Client>();
 362                 for( String role: Roles ) {
 363                         MR_Client client = new MR_Client();
 364                         client.setAction(actions);
 365                         client.setClientRole(role);
 366                         client.setDcaeLocationName( node.getDcaeLocationName());
 367                         clients.add( client );
 368                 }
 369                 bridgeAdminTopic.setClients(clients);
 370
 371                 TopicService ts = new TopicService();
 372                 ApiError err = new ApiError();
 373                 ts.addTopic(bridgeAdminTopic, err, true);
 374
 375                 if ( err.is2xx() || err.getCode() == 409 ){
 376                         err.setCode(200);
 377                         return err;
 378                 }
 379
 380                 errorLogger.error( DmaapbcLogMessageEnum.TOPIC_CREATE_ERROR,  bridgeAdminTopic.getFqtn(), Integer.toString(err.getCode()), err.getFields(), err.getMessage());
 381                 return err;
 382         }
 383
 384
 385         public Topic checkForBridge( Topic topic, ApiError err ) {
 386                 logger.info( "checkForBridge: entry");
 387                 logger.info( "fqtn=" + topic.getFqtn() + " replicatonType=" + topic.getReplicationCase());
 388                 if ( topic.getReplicationCase() == ReplicationType.REPLICATION_NONE ) {
 389                         topic.setStatus( DmaapObject_Status.VALID);
 390                         return topic;
 391                 }
 392
 393                 boolean anythingWrong = false;
 394
 395                 Set<String> groups = clusters.getGroups();
 396                 for ( String g : groups ) {
 397                         logger.info( "buildBridge for " + topic.getFqtn() + " on group" + g);
 398                         anythingWrong |= buildBridge( topic, err, g );
 399                 }
 400                 if ( anythingWrong ) {
 401                         topic.setStatus( DmaapObject_Status.INVALID);
 402                         if ( ! err.is2xx() ) {
 403                                 return null;
 404                         }
 405                 } else {
 406                         topic.setStatus( DmaapObject_Status.VALID);
 407                 }
 408                 return topic;
 409         }
 410
 411         private boolean buildBridge( Topic topic, ApiError err, String group ) {
 412                 logger.info( "buildBridge: entry");
 413                 boolean anythingWrong = false;
 414                 Graph graph;
 415                 logger.info( "buildBridge: strictGraph=" + strictGraph );
 416                 if ( group == null || group.isEmpty() ) {
 417                         graph = new Graph( topic.getClients(), strictGraph );
 418                 } else {
 419                         graph = new Graph( topic.getClients(), strictGraph, group );
 420                 }
 421                 logger.info( "buildBridge: graph=" + graph );
 422                 MR_Cluster groupCentralCluster = null;
 423
 424
 425                 if ( graph.isEmpty() ) {
 426                         logger.info( "buildBridge: graph is empty.  return false" );
 427                         return false;
 428                 } else if ( group == null &&  topic.getReplicationCase().involvesFQDN() ) {
 429                         logger.info( "buildBridge: group is null and replicationCaseInvolvesFQDN. return false" );
 430                         return false;
 431                 } else if ( ! graph.hasCentral() ) {
 432                         logger.warn( "Topic " + topic.getFqtn() + " wants to be " + topic.getReplicationCase() + " but has no central clients");
 433                         return true;
 434                 } else {
 435                         groupCentralCluster = clusters.getMr_ClusterByLoc(graph.getCentralLoc());
 436                 }
 437                 Collection<String> clientLocations = graph.getKeys();
 438                 for( String loc : clientLocations ) {
 439                         logger.info( "loc=" + loc );
 440                         DcaeLocation location = locations.getDcaeLocation(loc);
 441                         MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
 442                         logger.info( "cluster=" + cluster + " at "+ cluster.getDcaeLocationName() );
 443                         logger.info( "location.isCentral()="+location.isCentral() + " getCentralLoc()=" + graph.getCentralLoc() );
 444
 445
 446
 447                         String source = null;
 448                         String target = null;
 449
 450                         /*
 451                          * Provision Edge to Central bridges...
 452                          */
 453                         if ( ! location.isCentral()  && ! graph.getCentralLoc().equals(cluster.getDcaeLocationName()) ) {
 454                                 switch( topic.getReplicationCase() ) {
 455                                 case REPLICATION_EDGE_TO_CENTRAL:
 456                                 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:  // NOTE: this is for E2C portion only
 457                                         source = cluster.getFqdn();
 458                                         target = (mmPerMR)? groupCentralCluster.getFqdn() : centralCname;
 459                                         logger.info( "REPLICATION_EDGE_TO_CENTRAL: source=" + source + " target=" +target );
 460                                         break;
 461                                 case REPLICATION_CENTRAL_TO_EDGE:
 462                                 case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE:  // NOTE: this is for C2E portion only
 463                                         source = (mmPerMR) ? groupCentralCluster.getFqdn() : centralCname;
 464                                         target = cluster.getFqdn();
 465                                         break;
 466                                 case REPLICATION_CENTRAL_TO_GLOBAL:
 467                                 case REPLICATION_GLOBAL_TO_CENTRAL:
 468                                 case REPLICATION_FQDN_TO_GLOBAL:
 469                                 case REPLICATION_GLOBAL_TO_FQDN:
 470                                         break;
 471
 472                                 case REPLICATION_EDGE_TO_FQDN:
 473                                 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2C portion only
 474                                         source = cluster.getFqdn();
 475                                         target = groupCentralCluster.getFqdn();
 476                                         break;
 477                                 case REPLICATION_FQDN_TO_EDGE:
 478                                 case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE:  // NOTE: this is for F2E portion only
 479                                         source = groupCentralCluster.getFqdn();
 480                                         target = cluster.getFqdn();
 481                                         break;
 482
 483                                 default:
 484                                         logger.error( "Unexpected value for ReplicationType ("+ topic.getReplicationCase() + ") for topic " + topic.getFqtn() );
 485                                         anythingWrong = true;
 486                                         err.setCode(400);
 487                                         err.setFields("topic=" + topic.genFqtn() + " replicationCase="
 488                                                         + topic.getReplicationCase() );
 489                                         err.setMessage("Unexpected value for ReplicationType");
 490                                         continue;
 491                                 }
 492
 493                         } else if ( location.isCentral() && graph.getCentralLoc().equals(cluster.getDcaeLocationName()) ) {
 494                                 /*
 495                                  * Provision Central to Global bridges
 496                                  */
 497                                 switch( topic.getReplicationCase() ) {
 498
 499                                 case REPLICATION_CENTRAL_TO_GLOBAL:
 500                                 case REPLICATION_EDGE_TO_CENTRAL_TO_GLOBAL:
 501                                         source = centralCname;
 502                                         target = topic.getGlobalMrURL();
 503                                         break;
 504                                 case REPLICATION_GLOBAL_TO_CENTRAL:
 505                                 case REPLICATION_GLOBAL_TO_CENTRAL_TO_EDGE:  // NOTE: this is for G2C portion only
 506                                         source = topic.getGlobalMrURL();
 507                                         target = centralCname;
 508                                         break;
 509
 510                                 case REPLICATION_EDGE_TO_FQDN_TO_GLOBAL:  // NOTE: this is for E2F portion only
 511                                         source = groupCentralCluster.getFqdn();
 512                                         target = topic.getGlobalMrURL();
 513                                         break;
 514
 515                                 case REPLICATION_FQDN_TO_GLOBAL:
 516                                         source = groupCentralCluster.getFqdn();
 517                                         target = topic.getGlobalMrURL();
 518                                         break;
 519
 520                                 case REPLICATION_GLOBAL_TO_FQDN:
 521                                 case REPLICATION_GLOBAL_TO_FQDN_TO_EDGE:  // NOTE: this is for G2F portion only
 522                                         source = topic.getGlobalMrURL();
 523                                         target = groupCentralCluster.getFqdn();
 524                                         break;
 525
 526                                 case REPLICATION_FQDN_TO_EDGE:
 527                                 case REPLICATION_EDGE_TO_FQDN:
 528                                 case REPLICATION_EDGE_TO_CENTRAL:
 529                                 case REPLICATION_CENTRAL_TO_EDGE:
 530                                         break;
 531                                 default:
 532                                         logger.error( "Unexpected value for ReplicationType ("+ topic.getReplicationCase() + ") for topic " + topic.getFqtn() );
 533                                         anythingWrong = true;
 534                                         err.setCode(400);
 535                                         err.setFields("topic=" + topic.genFqtn() + " replicationCase="
 536                                                         + topic.getReplicationCase() );
 537                                         err.setMessage("Unexpected value for ReplicationType");
 538                                         continue;
 539                                 }
 540                         } else {
 541                                 logger.warn( "dcaeLocation " + loc + " is neither Edge nor Central so no mmagent provisioning was done");
 542                                 anythingWrong = true;
 543                                 continue;
 544                         }
 545                         if ( source != null && target != null ) {
 546                                 try {
 547                                         logger.info( "Create a MM from " + source + " to " + target );
 548                                         MirrorMaker mm = bridge.findNextMM( source, target, topic.getFqtn());
 549                                         mm.addTopic(topic.getFqtn());
 550                                         bridge.updateMirrorMaker(mm);
 551                                 } catch ( Exception ex ) {
 552                                         err.setCode(500);
 553                                         err.setFields( "mirror_maker.topic");
 554                                         err.setMessage("Unexpected condition: " + ex );
 555                                         anythingWrong = true;
 556                                         break;
 557                                 }
 558                         }
 559
 560
 561                 }
 562                 return  anythingWrong;
 563
 564         }
 565
 566
 567         /*
 568          * Prior to 1707, we only supported EDGE_TO_CENTRAL replication.
 569          * This was determined automatically based on presence of edge publishers and central subscribers.
 570          * The following method is a modification of that original logic, to preserve some backwards compatibility,
 571          * i.e. to be used when no ReplicationType is specified.
 572          */
 573         public ReplicationType reviewTopic( Topic topic ) {
 574
 575
 576                 if ( topic.getNumClients() > 1 ) {
 577                         Graph graph = new Graph( topic.getClients(), false );
 578
 579                         String centralFqdn = new String();
 580                         if ( graph.hasCentral() ) {
 581                                 DmaapConfig p = (DmaapConfig)DmaapConfig.getConfig();
 582                                 centralFqdn = p.getProperty("MR.CentralCname");
 583                         }
 584
 585                         Collection<String> locations = graph.getKeys();
 586                         for( String loc : locations ) {
 587                                 logger.info( "loc=" + loc );
 588                                 MR_Cluster cluster = clusters.getMr_ClusterByLoc(loc);
 589                                 if ( cluster == null ) {
 590                                         logger.info( "No MR cluster for location " + loc );
 591                                         continue;
 592                                 }
 593                                 if ( graph.hasCentral() &&  ! graph.getCentralLoc().equals(cluster.getDcaeLocationName())) {
 594                                         logger.info( "Detected case for EDGE_TO_CENTRAL from " + cluster.getFqdn() + " to " + centralFqdn );
 595                                         return ReplicationType.REPLICATION_EDGE_TO_CENTRAL;
 596
 597                                 }
 598
 599                         }
 600                 }
 601
 602                 return ReplicationType.REPLICATION_NONE;
 603         }
 604
 605 }