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