catalog-be/src/main/java/org/openecomp/sdc/be/components/distribution/engine/CambriaHandler.java

   1 /*-
   2  * ============LICENSE_START=======================================================
   3  * SDC
   4  * ================================================================================
   5  * Copyright (C) 2017 AT&T Intellectual Property. All rights reserved.
   6  * ================================================================================
   7  * Licensed under the Apache License, Version 2.0 (the "License");
   8  * you may not use this file except in compliance with the License.
   9  * You may obtain a copy of the License at
  10  *
  11  *      http://www.apache.org/licenses/LICENSE-2.0
  12  *
  13  * Unless required by applicable law or agreed to in writing, software
  14  * distributed under the License is distributed on an "AS IS" BASIS,
  15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  16  * See the License for the specific language governing permissions and
  17  * limitations under the License.
  18  * ============LICENSE_END=========================================================
  19  */
  20
  21 package org.openecomp.sdc.be.components.distribution.engine;
  22
  23 import java.io.IOException;
  24 import java.net.MalformedURLException;
  25 import java.security.GeneralSecurityException;
  26 import java.util.ArrayList;
  27 import java.util.Collection;
  28 import java.util.Iterator;
  29 import java.util.List;
  30 import java.util.Set;
  31 import java.util.concurrent.TimeUnit;
  32 import java.util.regex.Matcher;
  33 import java.util.regex.Pattern;
  34
  35 import org.apache.http.HttpStatus;
  36 import org.openecomp.sdc.be.config.BeEcompErrorManager;
  37 import org.openecomp.sdc.be.distribution.api.client.CambriaOperationStatus;
  38 import org.openecomp.sdc.common.config.EcompErrorName;
  39 import org.slf4j.Logger;
  40 import org.slf4j.LoggerFactory;
  41
  42 import com.att.nsa.apiClient.http.HttpException;
  43 import com.att.nsa.apiClient.http.HttpObjectNotFoundException;
  44 import com.att.nsa.cambria.client.CambriaBatchingPublisher;
  45 import com.att.nsa.cambria.client.CambriaClient.CambriaApiException;
  46 import com.att.nsa.cambria.client.CambriaClientBuilders.TopicManagerBuilder;
  47 import com.att.nsa.cambria.client.CambriaClientBuilders.PublisherBuilder;
  48 import com.att.nsa.cambria.client.CambriaClientBuilders.ConsumerBuilder;
  49 import com.att.nsa.cambria.client.CambriaClientBuilders.IdentityManagerBuilder;
  50 import com.att.nsa.cambria.client.CambriaConsumer;
  51 import com.att.nsa.cambria.client.CambriaIdentityManager;
  52 import com.att.nsa.cambria.client.CambriaPublisher.message;
  53 import com.att.nsa.cambria.client.CambriaTopicManager;
  54 import com.google.gson.Gson;
  55
  56 import fj.data.Either;
  57
  58 public class CambriaHandler {
  59
  60         private static Logger logger = LoggerFactory.getLogger(CambriaHandler.class.getName());
  61
  62         public static String PARTITION_KEY = "asdc" + "aa";
  63
  64         private Gson gson = new Gson();
  65
  66         /**
  67          * process the response error from Cambria client
  68          *
  69          * @param message
  70          * @return
  71          */
  72         private Integer processMessageException(String message) {
  73
  74                 String[] patterns = { "(HTTP Status )(\\d\\d\\d)", "(HTTP/\\d.\\d )(\\d\\d\\d)" };
  75
  76                 Integer result = checkPattern(patterns[0], message, 2);
  77                 if (result != null) {
  78                         return result;
  79                 }
  80                 result = checkPattern(patterns[1], message, 2);
  81
  82                 return result;
  83
  84         }
  85
  86         /**
  87          * check whether the message has a match with a given pattern inside it
  88          *
  89          * @param patternStr
  90          * @param message
  91          * @param groupIndex
  92          * @return
  93          */
  94         private Integer checkPattern(String patternStr, String message, int groupIndex) {
  95                 Integer result = null;
  96
  97                 Pattern pattern = Pattern.compile(patternStr);
  98                 Matcher matcher = pattern.matcher(message);
  99                 boolean find = matcher.find();
 100                 if (find) {
 101                         String httpCode = matcher.group(groupIndex);
 102                         if (httpCode != null) {
 103                                 try {
 104                                         result = Integer.valueOf(httpCode);
 105                                 } catch (NumberFormatException e) {
 106                                         logger.debug("Failed to parse http code {}", httpCode);
 107                                 }
 108                         }
 109                 }
 110                 return result;
 111         }
 112
 113         /**
 114          * retrieve all topics from U-EB server
 115          *
 116          * @param hostSet
 117          * @return
 118          */
 119         public Either<Set<String>, CambriaErrorResponse> getTopics(List<String> hostSet) {
 120
 121                 CambriaTopicManager createTopicManager = null;
 122                 try {
 123
 124                         createTopicManager = new TopicManagerBuilder().usingHosts(hostSet).build();
 125                         Set<String> topics = createTopicManager.getTopics();
 126
 127                         if (topics == null || true == topics.isEmpty()) {
 128                                 CambriaErrorResponse cambriaErrorResponse = new CambriaErrorResponse(CambriaOperationStatus.NOT_FOUND, null);
 129                                 return Either.right(cambriaErrorResponse);
 130                         }
 131
 132                         return Either.left(topics);
 133
 134                 } catch (IOException | GeneralSecurityException e) {
 135                         String methodName = new Object() {
 136                         }.getClass().getEnclosingMethod().getName();
 137
 138                         CambriaErrorResponse cambriaErrorResponse = processError(e);
 139
 140                         logger.debug("Failed to fetch topics from U-EB server", e);
 141                         writeErrorToLog(cambriaErrorResponse, e.getMessage(), methodName, "get topics");
 142
 143                         return Either.right(cambriaErrorResponse);
 144                 } finally {
 145                         if (createTopicManager != null) {
 146                                 createTopicManager.close();
 147                         }
 148                 }
 149
 150         }
 151
 152         /**
 153          * process the error message from Cambria client.
 154          *
 155          * set Cambria status and http code in case we succeed to fetch it
 156          *
 157          * @param errorMessage
 158          * @return
 159          */
 160         private CambriaErrorResponse processError(Exception e) {
 161
 162                 CambriaErrorResponse cambriaErrorResponse = new CambriaErrorResponse();
 163
 164                 Integer httpCode = processMessageException(e.getMessage());
 165
 166                 if (httpCode != null) {
 167                         cambriaErrorResponse.setHttpCode(httpCode);
 168                         switch (httpCode.intValue()) {
 169
 170                         case 401:
 171                                 cambriaErrorResponse.setOperationStatus(CambriaOperationStatus.AUTHENTICATION_ERROR);
 172                                 break;
 173                         case 409:
 174                                 cambriaErrorResponse.setOperationStatus(CambriaOperationStatus.TOPIC_ALREADY_EXIST);
 175                                 break;
 176                         case 500:
 177                                 cambriaErrorResponse.setOperationStatus(CambriaOperationStatus.INTERNAL_SERVER_ERROR);
 178                                 break;
 179                         default:
 180                                 cambriaErrorResponse.setOperationStatus(CambriaOperationStatus.CONNNECTION_ERROR);
 181                         }
 182                 } else {
 183
 184                         boolean found = false;
 185                         Throwable throwable = e.getCause();
 186                         if (throwable != null) {
 187                                 String message = throwable.getMessage();
 188
 189                                 Throwable cause = throwable.getCause();
 190
 191                                 if (cause != null) {
 192                                         Class<?> clazz = cause.getClass();
 193                                         String className = clazz.getName();
 194                                         if (className.endsWith("UnknownHostException")) {
 195                                                 cambriaErrorResponse.setOperationStatus(CambriaOperationStatus.UNKNOWN_HOST_ERROR);
 196                                                 cambriaErrorResponse.addVariable(message);
 197                                                 found = true;
 198                                         }
 199                                 }
 200                         }
 201
 202                         if (false == found) {
 203                                 cambriaErrorResponse.setOperationStatus(CambriaOperationStatus.CONNNECTION_ERROR);
 204                                 cambriaErrorResponse.setHttpCode(HttpStatus.SC_INTERNAL_SERVER_ERROR);
 205                         }
 206                 }
 207
 208                 return cambriaErrorResponse;
 209         }
 210
 211         /**
 212          * write the error to the log
 213          *
 214          * @param cambriaErrorResponse
 215          * @param errorMessage
 216          * @param methodName
 217          * @param operationDesc
 218          */
 219         private void writeErrorToLog(CambriaErrorResponse cambriaErrorResponse, String errorMessage, String methodName, String operationDesc) {
 220
 221                 String httpCode = (cambriaErrorResponse.getHttpCode() == null ? "" : String.valueOf(cambriaErrorResponse.getHttpCode()));
 222
 223                 switch (cambriaErrorResponse.getOperationStatus()) {
 224                 case UNKNOWN_HOST_ERROR:
 225                         String hostname = cambriaErrorResponse.getVariables().get(0);
 226                         BeEcompErrorManager.getInstance().processEcompError(EcompErrorName.BeUebUnkownHostError, methodName, hostname);
 227                         BeEcompErrorManager.getInstance().logBeUebUnkownHostError(methodName, httpCode);
 228                         break;
 229                 case AUTHENTICATION_ERROR:
 230                         BeEcompErrorManager.getInstance().processEcompError(EcompErrorName.BeUebAuthenticationError, methodName, httpCode);
 231                         BeEcompErrorManager.getInstance().logBeUebAuthenticationError(methodName, httpCode);
 232                         break;
 233                 case CONNNECTION_ERROR:
 234                         BeEcompErrorManager.getInstance().processEcompError(EcompErrorName.BeUebConnectionError, methodName, httpCode);
 235                         BeEcompErrorManager.getInstance().logBeUebConnectionError(methodName, httpCode);
 236                         break;
 237
 238                 case INTERNAL_SERVER_ERROR:
 239                         BeEcompErrorManager.getInstance().processEcompError(EcompErrorName.BeUebSystemError, methodName, operationDesc);
 240                         BeEcompErrorManager.getInstance().logBeUebSystemError(methodName, operationDesc);
 241                         break;
 242
 243                 }
 244
 245         }
 246
 247         /**
 248          * create a topic if it does not exists in the topicsList
 249          *
 250          * @param hostSet
 251          *            - list of U-EB servers
 252          * @param apiKey
 253          * @param secretKey
 254          * @param topicsList
 255          *            - list of exists topics
 256          * @param topicName
 257          *            - topic to create
 258          * @param partitionCount
 259          * @param replicationCount
 260          * @return
 261          */
 262         public CambriaErrorResponse createTopic(Collection<String> hostSet, String apiKey, String secretKey, String topicName, int partitionCount, int replicationCount) {
 263
 264                 CambriaTopicManager createTopicManager = null;
 265                 try {
 266                         createTopicManager = new TopicManagerBuilder().usingHosts(hostSet).authenticatedBy(apiKey, secretKey).build();
 267                         createTopicManager.createTopic(topicName, "ASDC distribution notification topic", partitionCount, replicationCount);
 268
 269                 } catch (GeneralSecurityException | HttpException | IOException e) {
 270                         logger.debug("Failed to create topic {}", topicName, e);
 271                         String methodName = new Object() {
 272                         }.getClass().getEnclosingMethod().getName();
 273
 274                         CambriaErrorResponse cambriaErrorResponse = processError(e);
 275
 276                         if (cambriaErrorResponse.getOperationStatus() != CambriaOperationStatus.TOPIC_ALREADY_EXIST) {
 277                                 writeErrorToLog(cambriaErrorResponse, e.getMessage(), methodName, "create topic");
 278                         }
 279
 280                         return cambriaErrorResponse;
 281
 282                 } finally {
 283                         if (createTopicManager != null) {
 284                                 createTopicManager.close();
 285                         }
 286                 }
 287                 return new CambriaErrorResponse(CambriaOperationStatus.OK);
 288         }
 289
 290         public CambriaErrorResponse unRegisterFromTopic(Collection<String> hostSet, String topicName, String managerApiKey, String managerSecretKey, String subscriberApiKey, SubscriberTypeEnum subscriberTypeEnum) {
 291                 CambriaTopicManager createTopicManager = null;
 292                 try {
 293                         createTopicManager = new TopicManagerBuilder().usingHosts(hostSet).authenticatedBy(managerApiKey, managerSecretKey).build();
 294
 295                         if (subscriberTypeEnum == SubscriberTypeEnum.PRODUCER) {
 296                                 createTopicManager.revokeProducer(topicName, subscriberApiKey);
 297                         } else {
 298                                 createTopicManager.revokeConsumer(topicName, subscriberApiKey);
 299                         }
 300
 301                 } catch (HttpObjectNotFoundException | GeneralSecurityException e) {
 302                         logger.debug("Failed to unregister {} from topic {} as {}", managerApiKey, topicName, subscriberTypeEnum.toString().toLowerCase(), e);
 303                         String methodName = new Object() {
 304                         }.getClass().getEnclosingMethod().getName();
 305
 306                         BeEcompErrorManager.getInstance().processEcompError(EcompErrorName.BeUebObjectNotFoundError, methodName, e.getMessage());
 307                         BeEcompErrorManager.getInstance().logBeUebObjectNotFoundError(methodName, e.getMessage());
 308
 309                         CambriaErrorResponse cambriaErrorResponse = new CambriaErrorResponse(CambriaOperationStatus.OBJECT_NOT_FOUND, HttpStatus.SC_NOT_FOUND);
 310                         return cambriaErrorResponse;
 311
 312                 } catch (HttpException | IOException e) {
 313                         logger.debug("Failed to unregister {} from topic {} as producer", managerApiKey, topicName, e);
 314                         String methodName = new Object() {
 315                         }.getClass().getEnclosingMethod().getName();
 316
 317                         CambriaErrorResponse cambriaErrorResponse = processError(e);
 318
 319                         writeErrorToLog(cambriaErrorResponse, e.getMessage(), methodName, "unregister from topic as " + subscriberTypeEnum.toString().toLowerCase());
 320
 321                         return cambriaErrorResponse;
 322                 } finally {
 323                         if (createTopicManager != null) {
 324                                 createTopicManager.close();
 325                         }
 326                 }
 327
 328                 CambriaErrorResponse cambriaErrorResponse = new CambriaErrorResponse(CambriaOperationStatus.OK, HttpStatus.SC_OK);
 329                 return cambriaErrorResponse;
 330         }
 331
 332         /**
 333          *
 334          * register a public key (subscriberId) to a given topic as a CONSUMER or PRODUCER
 335          *
 336          * @param hostSet
 337          * @param topicName
 338          * @param managerApiKey
 339          * @param managerSecretKey
 340          * @param subscriberApiKey
 341          * @param subscriberTypeEnum
 342          * @return
 343          */
 344         public CambriaErrorResponse registerToTopic(Collection<String> hostSet, String topicName, String managerApiKey, String managerSecretKey, String subscriberApiKey, SubscriberTypeEnum subscriberTypeEnum) {
 345
 346                 CambriaTopicManager createTopicManager = null;
 347                 try {
 348                         createTopicManager = new TopicManagerBuilder().usingHosts(hostSet).authenticatedBy(managerApiKey, managerSecretKey).build();
 349
 350                         if (subscriberTypeEnum == SubscriberTypeEnum.PRODUCER) {
 351                                 createTopicManager.allowProducer(topicName, subscriberApiKey);
 352                         } else {
 353                                 createTopicManager.allowConsumer(topicName, subscriberApiKey);
 354                         }
 355
 356                 } catch (HttpObjectNotFoundException | GeneralSecurityException e) {
 357                         logger.debug("Failed to register {} to topic {} as {}", managerApiKey, topicName, subscriberTypeEnum.toString().toLowerCase(), e);
 358                         String methodName = new Object() {
 359                         }.getClass().getEnclosingMethod().getName();
 360
 361                         BeEcompErrorManager.getInstance().processEcompError(EcompErrorName.BeUebObjectNotFoundError, methodName, e.getMessage());
 362                         BeEcompErrorManager.getInstance().logBeUebObjectNotFoundError(methodName, e.getMessage());
 363
 364                         CambriaErrorResponse cambriaErrorResponse = new CambriaErrorResponse(CambriaOperationStatus.OBJECT_NOT_FOUND, HttpStatus.SC_NOT_FOUND);
 365                         return cambriaErrorResponse;
 366
 367                 } catch (HttpException | IOException e) {
 368                         logger.debug("Failed to register {} to topic {} as {}", managerApiKey, topicName, subscriberTypeEnum.toString().toLowerCase(), e);
 369                         String methodName = new Object() {
 370                         }.getClass().getEnclosingMethod().getName();
 371
 372                         CambriaErrorResponse cambriaErrorResponse = processError(e);
 373
 374                         writeErrorToLog(cambriaErrorResponse, e.getMessage(), methodName, "register to topic as " + subscriberTypeEnum.toString().toLowerCase());
 375
 376                         return cambriaErrorResponse;
 377                 } finally {
 378                         if (createTopicManager != null) {
 379                                 createTopicManager.close();
 380                         }
 381                 }
 382
 383                 CambriaErrorResponse cambriaErrorResponse = new CambriaErrorResponse(CambriaOperationStatus.OK, HttpStatus.SC_OK);
 384                 return cambriaErrorResponse;
 385         }
 386
 387         /**
 388          * create and retrieve a Cambria Consumer for a specific topic
 389          *
 390          * @param hostSet
 391          * @param topicName
 392          * @param apiKey
 393          * @param secretKey
 394          * @param consumerId
 395          * @param consumerGroup
 396          * @param timeoutMS
 397          * @return
 398          * @throws Exception
 399          */
 400         public CambriaConsumer createConsumer(Collection<String> hostSet, String topicName, String apiKey, String secretKey, String consumerId, String consumerGroup, int timeoutMS) throws Exception {
 401
 402                 CambriaConsumer consumer = new ConsumerBuilder().authenticatedBy(apiKey, secretKey).knownAs(consumerGroup, consumerId).onTopic(topicName).usingHosts(hostSet).withSocketTimeout(timeoutMS).build();
 403                 consumer.setApiCredentials(apiKey, secretKey);
 404                 return consumer;
 405         }
 406
 407         public void closeConsumer(CambriaConsumer consumer) {
 408
 409                 if (consumer != null) {
 410                         consumer.close();
 411                 }
 412
 413         }
 414
 415         /**
 416          * use the topicConsumer to fetch messages from topic. in case no messages were fetched, empty ArrayList will be returned (not error)
 417          *
 418          * @param topicConsumer
 419          * @return
 420          */
 421         public Either<Iterable<String>, CambriaErrorResponse> fetchFromTopic(CambriaConsumer topicConsumer) {
 422
 423                 try {
 424                         Iterable<String> messages = topicConsumer.fetch();
 425                         if (messages == null) {
 426                                 messages = new ArrayList<String>();
 427                         }
 428                         return Either.left(messages);
 429
 430                 } catch (IOException e) {
 431                         String methodName = new Object() {
 432                         }.getClass().getEnclosingMethod().getName();
 433
 434                         CambriaErrorResponse cambriaErrorResponse = processError(e);
 435
 436                         logger.debug("Failed to fetch from U-EB topic. error={}", e.getMessage());
 437                         writeErrorToLog(cambriaErrorResponse, e.getMessage(), methodName, "get messages from topic");
 438
 439                         return Either.right(cambriaErrorResponse);
 440
 441                 } catch (Exception e) {
 442                         logger.debug("Failed to fetch from U-EB topic", e);
 443                         String methodName = new Object() {
 444                         }.getClass().getEnclosingMethod().getName();
 445
 446                         BeEcompErrorManager.getInstance().processEcompError(EcompErrorName.BeDistributionEngineSystemError, methodName, e.getMessage());
 447                         BeEcompErrorManager.getInstance().logBeDistributionEngineSystemError(methodName, e.getMessage());
 448
 449                         CambriaErrorResponse cambriaErrorResponse = new CambriaErrorResponse(CambriaOperationStatus.INTERNAL_SERVER_ERROR, HttpStatus.SC_INTERNAL_SERVER_ERROR);
 450                         return Either.right(cambriaErrorResponse);
 451                 }
 452         }
 453
 454         /**
 455          * Publish notification message to a given queue
 456          *
 457          * @param topicName
 458          * @param uebPublicKey
 459          * @param uebSecretKey
 460          * @param uebServers
 461          * @param data
 462          * @return
 463          */
 464         public CambriaErrorResponse sendNotification(String topicName, String uebPublicKey, String uebSecretKey, List<String> uebServers, INotificationData data) {
 465
 466                 CambriaBatchingPublisher createSimplePublisher = null;
 467
 468                 try {
 469
 470                         String json = gson.toJson(data);
 471                         logger.trace("Before sending notification data {} to topic {}", json, topicName);
 472
 473                         createSimplePublisher = new PublisherBuilder().onTopic(topicName).usingHosts(uebServers).build();
 474                         createSimplePublisher.setApiCredentials(uebPublicKey, uebSecretKey);
 475
 476                         int result = createSimplePublisher.send(PARTITION_KEY, json);
 477
 478                         try {
 479                                 Thread.sleep(1 * 1000);
 480                         } catch (InterruptedException e) {
 481                                 logger.debug("Failed during sleep after sending the message.", e);
 482                         }
 483
 484                         logger.debug("After sending notification data to topic {}. result is {}", topicName, result);
 485
 486                         CambriaErrorResponse response = new CambriaErrorResponse(CambriaOperationStatus.OK, 200);
 487
 488                         return response;
 489
 490                 } catch (IOException | GeneralSecurityException e) {
 491                         logger.debug("Failed to send notification {} to topic {} ", data, topicName, e);
 492
 493                         String methodName = new Object() {
 494                         }.getClass().getEnclosingMethod().getName();
 495
 496                         CambriaErrorResponse cambriaErrorResponse = processError(e);
 497
 498                         writeErrorToLog(cambriaErrorResponse, e.getMessage(), methodName, "send notification");
 499
 500                         return cambriaErrorResponse;
 501                 } finally {
 502                         if (createSimplePublisher != null) {
 503                                 logger.debug("Before closing publisher");
 504                                 createSimplePublisher.close();
 505                                 logger.debug("After closing publisher");
 506                         }
 507                 }
 508         }
 509
 510         private String convertListToString(List<String> list) {
 511                 StringBuilder builder = new StringBuilder();
 512
 513                 if (list != null) {
 514                         for (int i = 0; i < list.size(); i++) {
 515                                 builder.append(list.get(i));
 516                                 if (i < list.size() - 1) {
 517                                         builder.append(",");
 518                                 }
 519                         }
 520                 }
 521
 522                 return builder.toString();
 523         }
 524
 525         public CambriaErrorResponse sendNotificationAndClose(String topicName, String uebPublicKey, String uebSecretKey, List<String> uebServers, INotificationData data, long waitBeforeCloseTimeout) {
 526
 527                 CambriaBatchingPublisher createSimplePublisher = null;
 528
 529                 CambriaErrorResponse response = null;
 530                 try {
 531
 532                         String json = gson.toJson(data);
 533                         logger.debug("Before sending notification data {} to topic {}", json, topicName);
 534
 535                         createSimplePublisher = new PublisherBuilder().onTopic(topicName).usingHosts(uebServers).build();
 536                         createSimplePublisher.setApiCredentials(uebPublicKey, uebSecretKey);
 537
 538                         int result = createSimplePublisher.send(PARTITION_KEY, json);
 539
 540                         try {
 541                                 Thread.sleep(1000);
 542                         } catch (InterruptedException e) {
 543                                 logger.debug("Failed during sleep after sending the message.", e);
 544                         }
 545
 546                         logger.debug("After sending notification data to topic {}. result is {}", topicName, result);
 547
 548                 } catch (IOException | GeneralSecurityException e) {
 549                         logger.debug("Failed to send notification {} to topic {} ", data, topicName, e);
 550
 551                         String methodName = new Object() {
 552                         }.getClass().getEnclosingMethod().getName();
 553
 554                         response = processError(e);
 555
 556                         writeErrorToLog(response, e.getMessage(), methodName, "send notification");
 557
 558                         return response;
 559
 560                 }
 561
 562                 logger.debug("Before closing publisher. Maximum timeout is {} seconds.", waitBeforeCloseTimeout);
 563                 try {
 564                         List<message> messagesInQ = createSimplePublisher.close(waitBeforeCloseTimeout, TimeUnit.SECONDS);
 565                         if (messagesInQ != null && false == messagesInQ.isEmpty()) {
 566                                 logger.debug("Cambria client returned {} non sent messages.", messagesInQ.size());
 567                                 response = new CambriaErrorResponse(CambriaOperationStatus.INTERNAL_SERVER_ERROR, 500);
 568                                 String methodName = new Object() {
 569                                 }.getClass().getEnclosingMethod().getName();
 570                                 writeErrorToLog(response, "closing publisher returned non sent messages", methodName, "send notification");
 571                         } else {
 572                                 logger.debug("No message left in the queue after closing cambria publisher");
 573                                 response = new CambriaErrorResponse(CambriaOperationStatus.OK, 200);
 574                         }
 575                 } catch (IOException | InterruptedException e) {
 576                         logger.debug("Failed to close cambria publisher", e);
 577                         response = new CambriaErrorResponse(CambriaOperationStatus.INTERNAL_SERVER_ERROR, 500);
 578                         String methodName = new Object() {
 579                         }.getClass().getEnclosingMethod().getName();
 580                         writeErrorToLog(response, "closing publisher returned non sent messages", methodName, "send notification");
 581                 }
 582                 logger.debug("After closing publisher");
 583
 584                 return response;
 585
 586         }
 587
 588         public CambriaErrorResponse getApiKey(String server, String apiKey) {
 589
 590                 CambriaErrorResponse response = null;
 591
 592                 List<String> hostSet = new ArrayList<>();
 593                 hostSet.add(server);
 594                 CambriaIdentityManager createIdentityManager = null;
 595                 try {
 596                         createIdentityManager = new IdentityManagerBuilder().usingHosts(hostSet).build();
 597                         createIdentityManager.getApiKey(apiKey);
 598                         response = new CambriaErrorResponse(CambriaOperationStatus.OK, 200);
 599
 600                 } catch (HttpException | IOException | CambriaApiException | GeneralSecurityException e) {
 601                         logger.debug("Failed to fetch api key {} from server ", apiKey, server, e);
 602
 603                         response = processError(e);
 604
 605                 }
 606
 607                 return response;
 608         }
 609
 610 }