src/main/java/org/openecomp/sparky/synchronizer/AutosuggestionSynchronizer.java

   1 /**
   2  * ============LICENSE_START===================================================
   3  * SPARKY (AAI UI service)
   4  * ============================================================================
   5  * Copyright © 2017 AT&T Intellectual Property.
   6  * Copyright © 2017 Amdocs
   7  * All rights reserved.
   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  * ECOMP and OpenECOMP are trademarks
  23  * and service marks of AT&T Intellectual Property.
  24  */
  25
  26 package org.openecomp.sparky.synchronizer;
  27
  28 import static java.util.concurrent.CompletableFuture.supplyAsync;
  29
  30 import java.io.IOException;
  31 import java.util.ArrayList;
  32 import java.util.Arrays;
  33 import java.util.Collection;
  34 import java.util.Deque;
  35 import java.util.EnumSet;
  36 import java.util.HashMap;
  37 import java.util.Iterator;
  38 import java.util.List;
  39 import java.util.Map;
  40 import java.util.concurrent.ConcurrentHashMap;
  41 import java.util.concurrent.ConcurrentLinkedDeque;
  42 import java.util.concurrent.ExecutorService;
  43 import java.util.concurrent.atomic.AtomicInteger;
  44 import java.util.function.Supplier;
  45
  46 import org.openecomp.cl.api.Logger;
  47 import org.openecomp.cl.eelf.LoggerFactory;
  48 import org.openecomp.cl.mdc.MdcContext;
  49 import org.openecomp.sparky.config.oxm.OxmEntityDescriptor;
  50 import org.openecomp.sparky.dal.NetworkTransaction;
  51 import org.openecomp.sparky.dal.aai.config.ActiveInventoryConfig;
  52 import org.openecomp.sparky.dal.rest.HttpMethod;
  53 import org.openecomp.sparky.dal.rest.OperationResult;
  54 import org.openecomp.sparky.logging.AaiUiMsgs;
  55 import org.openecomp.sparky.synchronizer.config.SynchronizerConfiguration;
  56 import org.openecomp.sparky.synchronizer.entity.SelfLinkDescriptor;
  57 import org.openecomp.sparky.synchronizer.entity.SuggestionSearchEntity;
  58 import org.openecomp.sparky.synchronizer.enumeration.OperationState;
  59 import org.openecomp.sparky.synchronizer.enumeration.SynchronizerState;
  60 import org.openecomp.sparky.synchronizer.task.PerformActiveInventoryRetrieval;
  61 import org.openecomp.sparky.synchronizer.task.PerformElasticSearchPut;
  62 import org.openecomp.sparky.synchronizer.task.PerformElasticSearchRetrieval;
  63 import org.openecomp.sparky.util.NodeUtils;
  64 import org.openecomp.sparky.util.SuggestionsPermutation;
  65 import org.slf4j.MDC;
  66
  67 import com.fasterxml.jackson.core.JsonProcessingException;
  68 import com.fasterxml.jackson.databind.JsonNode;
  69 import com.fasterxml.jackson.databind.node.ArrayNode;
  70
  71 /**
  72  * The Class AutosuggestionSynchronizer.
  73  */
  74 public class AutosuggestionSynchronizer extends AbstractEntitySynchronizer
  75     implements IndexSynchronizer {
  76
  77   private class RetrySuggestionEntitySyncContainer {
  78     NetworkTransaction txn;
  79     SuggestionSearchEntity ssec;
  80
  81     /**
  82      * Instantiates a new RetrySuggestionEntitySyncContainer.
  83      *
  84      * @param txn the txn
  85      * @param icer the icer
  86      */
  87     public RetrySuggestionEntitySyncContainer(NetworkTransaction txn, SuggestionSearchEntity icer) {
  88       this.txn = txn;
  89       this.ssec = icer;
  90     }
  91
  92     public NetworkTransaction getNetworkTransaction() {
  93       return txn;
  94     }
  95
  96     public SuggestionSearchEntity getSuggestionSearchEntity() {
  97       return ssec;
  98     }
  99   }
 100
 101   private static final Logger LOG =
 102       LoggerFactory.getInstance().getLogger(AutosuggestionSynchronizer.class);
 103   private static final String INSERTION_DATE_TIME_FORMAT = "yyyyMMdd'T'HHmmssZ";
 104
 105   private boolean allWorkEnumerated;
 106   private Deque<SelfLinkDescriptor> selflinks;
 107   private ConcurrentHashMap<String, AtomicInteger> entityCounters;
 108   private boolean syncInProgress;
 109   private Map<String, String> contextMap;
 110   protected ExecutorService esPutExecutor;
 111   private Deque<RetrySuggestionEntitySyncContainer> retryQueue;
 112   private Map<String, Integer> retryLimitTracker;
 113
 114   /**
 115    * Instantiates a new historical entity summarizer.
 116    *
 117    * @param indexName the index name
 118    * @throws Exception the exception
 119    */
 120   public AutosuggestionSynchronizer(String indexName) throws Exception {
 121     super(LOG, "ASES-" + indexName.toUpperCase(), 2, 5, 5, indexName); // multiple Autosuggestion
 122                                                                        // Entity Synchronizer will
 123                                                                        // run for different indices
 124
 125     this.allWorkEnumerated = false;
 126     this.selflinks = new ConcurrentLinkedDeque<SelfLinkDescriptor>();
 127     this.entityCounters = new ConcurrentHashMap<String, AtomicInteger>();
 128     this.synchronizerName = "Autosuggestion Entity Synchronizer";
 129     this.enabledStatFlags = EnumSet.of(StatFlag.AAI_REST_STATS, StatFlag.ES_REST_STATS);
 130     this.syncInProgress = false;
 131     this.contextMap = MDC.getCopyOfContextMap();
 132     this.esPutExecutor = NodeUtils.createNamedExecutor("SUES-ES-PUT", 5, LOG);
 133     this.syncDurationInMs = -1;
 134   }
 135
 136   /**
 137    * Collect all the work.
 138    *
 139    * @return the operation state
 140    */
 141   private OperationState collectAllTheWork() {
 142     final Map<String, String> contextMap = MDC.getCopyOfContextMap();
 143     Map<String, OxmEntityDescriptor> descriptorMap =
 144         oxmModelLoader.getSuggestionSearchEntityDescriptors();
 145
 146     if (descriptorMap.isEmpty()) {
 147       LOG.error(AaiUiMsgs.ERROR_LOADING_OXM_SUGGESTIBLE_ENTITIES);
 148       LOG.info(AaiUiMsgs.ERROR_LOADING_OXM_SUGGESTIBLE_ENTITIES);
 149       return OperationState.ERROR;
 150     }
 151
 152     Collection<String> syncTypes = descriptorMap.keySet();
 153
 154     try {
 155
 156       /*
 157        * launch a parallel async thread to process the documents for each entity-type (to max the of
 158        * the configured executor anyway)
 159        */
 160
 161       aaiWorkOnHand.set(syncTypes.size());
 162
 163       for (String key : syncTypes) {
 164
 165         supplyAsync(new Supplier<Void>() {
 166
 167           @Override
 168           public Void get() {
 169             MDC.setContextMap(contextMap);
 170             OperationResult typeLinksResult = null;
 171             try {
 172               typeLinksResult = aaiDataProvider.getSelfLinksByEntityType(key);
 173               aaiWorkOnHand.decrementAndGet();
 174               processEntityTypeSelfLinks(typeLinksResult);
 175             } catch (Exception exc) {
 176               // TODO -> LOG, what should be logged here?
 177             }
 178
 179             return null;
 180           }
 181
 182         }, aaiExecutor).whenComplete((result, error) -> {
 183
 184           if (error != null) {
 185             LOG.error(AaiUiMsgs.ERROR_GENERIC,
 186                 "An error occurred getting data from AAI. Error = " + error.getMessage());
 187           }
 188         });
 189
 190       }
 191
 192       while (aaiWorkOnHand.get() != 0) {
 193
 194         if (LOG.isDebugEnabled()) {
 195           LOG.debug(AaiUiMsgs.WAIT_FOR_ALL_SELFLINKS_TO_BE_COLLECTED);
 196         }
 197
 198         Thread.sleep(1000);
 199       }
 200
 201       aaiWorkOnHand.set(selflinks.size());
 202       allWorkEnumerated = true;
 203       syncEntityTypes();
 204
 205       while (!isSyncDone()) {
 206         performRetrySync();
 207         Thread.sleep(1000);
 208       }
 209
 210       /*
 211        * Make sure we don't hang on to retries that failed which could cause issues during future
 212        * syncs
 213        */
 214       retryLimitTracker.clear();
 215
 216     } catch (Exception exc) {
 217       // TODO -> LOG, waht should be logged here?
 218     }
 219
 220     return OperationState.OK;
 221
 222   }
 223
 224   /*
 225    * (non-Javadoc)
 226    *
 227    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#doSync()
 228    */
 229   @Override
 230   public OperationState doSync() {
 231         this.syncDurationInMs = -1;
 232         syncStartedTimeStampInMs = System.currentTimeMillis();
 233     String txnID = NodeUtils.getRandomTxnId();
 234     MdcContext.initialize(txnID, "AutosuggestionSynchronizer", "", "Sync", "");
 235
 236     return collectAllTheWork();
 237   }
 238
 239   /**
 240    * Process entity type self links.
 241    *
 242    * @param operationResult the operation result
 243    */
 244   private void processEntityTypeSelfLinks(OperationResult operationResult) {
 245
 246     JsonNode rootNode = null;
 247
 248     final String jsonResult = operationResult.getResult();
 249
 250     if (jsonResult != null && jsonResult.length() > 0 && operationResult.wasSuccessful()) {
 251
 252       try {
 253         rootNode = mapper.readTree(jsonResult);
 254       } catch (IOException exc) {
 255         String message = "Could not deserialize JSON (representing operation result) as node tree. "
 256             + "Operation result = " + jsonResult + ". " + exc.getLocalizedMessage();
 257         LOG.error(AaiUiMsgs.JSON_PROCESSING_ERROR, message);
 258       }
 259
 260       JsonNode resultData = rootNode.get("result-data");
 261       ArrayNode resultDataArrayNode = null;
 262
 263       if (resultData.isArray()) {
 264         resultDataArrayNode = (ArrayNode) resultData;
 265
 266         Iterator<JsonNode> elementIterator = resultDataArrayNode.elements();
 267         JsonNode element = null;
 268
 269         while (elementIterator.hasNext()) {
 270           element = elementIterator.next();
 271
 272           final String resourceType = NodeUtils.getNodeFieldAsText(element, "resource-type");
 273           final String resourceLink = NodeUtils.getNodeFieldAsText(element, "resource-link");
 274
 275           OxmEntityDescriptor descriptor = null;
 276
 277           if (resourceType != null && resourceLink != null) {
 278
 279             descriptor = oxmModelLoader.getEntityDescriptor(resourceType);
 280
 281             if (descriptor == null) {
 282               LOG.error(AaiUiMsgs.MISSING_ENTITY_DESCRIPTOR, resourceType);
 283               // go to next element in iterator
 284               continue;
 285             }
 286             selflinks.add(new SelfLinkDescriptor(resourceLink,
 287                 SynchronizerConfiguration.NODES_ONLY_MODIFIER, resourceType));
 288
 289
 290           }
 291         }
 292       }
 293     }
 294   }
 295
 296   /**
 297    * Sync entity types.
 298    */
 299   private void syncEntityTypes() {
 300
 301     while (selflinks.peek() != null) {
 302
 303       SelfLinkDescriptor linkDescriptor = selflinks.poll();
 304       aaiWorkOnHand.decrementAndGet();
 305
 306       OxmEntityDescriptor descriptor = null;
 307
 308       if (linkDescriptor.getSelfLink() != null && linkDescriptor.getEntityType() != null) {
 309
 310         descriptor = oxmModelLoader.getEntityDescriptor(linkDescriptor.getEntityType());
 311
 312         if (descriptor == null) {
 313           LOG.error(AaiUiMsgs.MISSING_ENTITY_DESCRIPTOR, linkDescriptor.getEntityType());
 314           // go to next element in iterator
 315           continue;
 316         }
 317
 318         NetworkTransaction txn = new NetworkTransaction();
 319         txn.setDescriptor(descriptor);
 320         txn.setLink(linkDescriptor.getSelfLink());
 321         txn.setOperationType(HttpMethod.GET);
 322         txn.setEntityType(linkDescriptor.getEntityType());
 323
 324         aaiWorkOnHand.incrementAndGet();
 325
 326         supplyAsync(new PerformActiveInventoryRetrieval(txn, aaiDataProvider), aaiExecutor)
 327             .whenComplete((result, error) -> {
 328
 329               aaiWorkOnHand.decrementAndGet();
 330
 331               if (error != null) {
 332                 LOG.error(AaiUiMsgs.AAI_RETRIEVAL_FAILED_GENERIC, error.getLocalizedMessage());
 333               } else {
 334                 if (result == null) {
 335                   LOG.error(AaiUiMsgs.AAI_RETRIEVAL_FAILED_FOR_SELF_LINK,
 336                       linkDescriptor.getSelfLink());
 337                 } else {
 338                   updateActiveInventoryCounters(result);
 339                   fetchDocumentForUpsert(result);
 340                 }
 341               }
 342             });
 343       }
 344
 345     }
 346
 347   }
 348   /*
 349    * Return a set of valid suggestion attributes for the provided entityName
 350    * that are present in the JSON
 351    * @param node    JSON node in which the attributes should be found
 352    * @param entityName  Name of the entity
 353    * @return List of all valid suggestion attributes(key's)
 354    */
 355   public List<String> getSuggestionFromReponse(JsonNode node, String entityName) {
 356     List<String> suggestableAttr = new ArrayList<String>();
 357     HashMap<String, String> desc = oxmModelLoader.getOxmModel().get(entityName);
 358     String attr = desc.get("suggestibleAttributes");
 359     suggestableAttr = Arrays.asList(attr.split(","));
 360     List<String> suggestableValue = new ArrayList<>();
 361     for (String attribute : suggestableAttr) {
 362       if (node.get(attribute) != null && node.get(attribute).asText().length() > 0) {
 363         suggestableValue.add(attribute);
 364       }
 365     }
 366     return suggestableValue;
 367   }
 368
 369   /**
 370    * Fetch all the documents for upsert. Based on the number of permutations that are available the
 371    * number of documents will be different
 372    *
 373    * @param txn the txn
 374    */
 375   private void fetchDocumentForUpsert(NetworkTransaction txn) {
 376     if (!txn.getOperationResult().wasSuccessful()) {
 377       String message = "Self link failure. Result - " + txn.getOperationResult().getResult();
 378       LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
 379       return;
 380     }
 381     try {
 382       final String jsonResult = txn.getOperationResult().getResult();
 383
 384       if (jsonResult != null && jsonResult.length() > 0) {
 385
 386         // Step 1: Calculate the number of possible permutations of attributes
 387         String entityName = txn.getDescriptor().getEntityName();
 388         JsonNode entityNode = mapper.readTree(jsonResult);
 389
 390         SuggestionsPermutation suggPermutation = new SuggestionsPermutation();
 391         ArrayList<ArrayList<String>> uniqueLists = suggPermutation
 392             .getSuggestionsPermutation(getSuggestionFromReponse(entityNode, entityName));
 393
 394         // Now we have a list of all possible permutations for the status that are
 395         // defined for this entity type. Try inserting a document for every combination.
 396         for (ArrayList<String> uniqueList : uniqueLists) {
 397           SuggestionSearchEntity sse = new SuggestionSearchEntity(oxmModelLoader);
 398           sse.setSuggestableAttr(uniqueList);
 399           sse.setPayloadFromResponse(entityNode);
 400           sse.setLink(txn.getLink());
 401           sse.setLink(ActiveInventoryConfig.extractResourcePath(txn.getLink()));
 402           populateSuggestionSearchEntityDocument(sse, jsonResult, txn);
 403           // The unique id for the document will be created at derive fields
 404           sse.deriveFields();
 405           // Insert the document only if it has valid statuses
 406           if (sse.isSuggestableDoc()) {
 407             String link = null;
 408             try {
 409               link = getElasticFullUrl("/" + sse.getId(), getIndexName());
 410             } catch (Exception exc) {
 411               LOG.error(AaiUiMsgs.ES_FAILED_TO_CONSTRUCT_QUERY, exc.getLocalizedMessage());
 412             }
 413
 414             if (link != null) {
 415               NetworkTransaction n2 = new NetworkTransaction();
 416               n2.setLink(link);
 417               n2.setEntityType(txn.getEntityType());
 418               n2.setDescriptor(txn.getDescriptor());
 419               n2.setOperationType(HttpMethod.GET);
 420
 421               esWorkOnHand.incrementAndGet();
 422
 423               supplyAsync(new PerformElasticSearchRetrieval(n2, esDataProvider), esExecutor)
 424                   .whenComplete((result, error) -> {
 425
 426                     esWorkOnHand.decrementAndGet();
 427
 428                     if (error != null) {
 429                       LOG.error(AaiUiMsgs.ES_RETRIEVAL_FAILED, error.getLocalizedMessage());
 430                     } else {
 431                       updateElasticSearchCounters(result);
 432                       performDocumentUpsert(result, sse);
 433                     }
 434                   });
 435             }
 436           }
 437         }
 438       }
 439     } catch (JsonProcessingException exc) {
 440       // TODO -> LOG, waht should be logged here?
 441     } catch (IOException exc) {
 442       // TODO -> LOG, waht should be logged here?
 443     }
 444   }
 445
 446   protected void populateSuggestionSearchEntityDocument(SuggestionSearchEntity sse, String result,
 447       NetworkTransaction txn) throws JsonProcessingException, IOException {
 448
 449     OxmEntityDescriptor resultDescriptor = txn.getDescriptor();
 450
 451     sse.setEntityType(resultDescriptor.getEntityName());
 452
 453     JsonNode entityNode = mapper.readTree(result);
 454
 455     List<String> primaryKeyValues = new ArrayList<String>();
 456     String pkeyValue = null;
 457
 458     for (String keyName : resultDescriptor.getPrimaryKeyAttributeName()) {
 459       pkeyValue = NodeUtils.getNodeFieldAsText(entityNode, keyName);
 460       if (pkeyValue != null) {
 461         primaryKeyValues.add(pkeyValue);
 462       } else {
 463         String message = "populateSuggestionSearchEntityDocument(),"
 464             + " pKeyValue is null for entityType = " + resultDescriptor.getEntityName();
 465         LOG.warn(AaiUiMsgs.WARN_GENERIC, message);
 466       }
 467     }
 468
 469     final String primaryCompositeKeyValue = NodeUtils.concatArray(primaryKeyValues, "/");
 470     sse.setEntityPrimaryKeyValue(primaryCompositeKeyValue);
 471     sse.generateSuggestionInputPermutations();
 472   }
 473
 474   protected void performDocumentUpsert(NetworkTransaction esGetTxn, SuggestionSearchEntity sse) {
 475     /**
 476      * <p>
 477      * <ul>
 478      * As part of the response processing we need to do the following:
 479      * <li>1. Extract the version (if present), it will be the ETAG when we use the
 480      * Search-Abstraction-Service
 481      * <li>2. Spawn next task which is to do the PUT operation into elastic with or with the version
 482      * tag
 483      * <li>a) if version is null or RC=404, then standard put, no _update with version tag
 484      * <li>b) if version != null, do PUT with _update?version= versionNumber in the URI to elastic
 485      * </ul>
 486      * </p>
 487      */
 488     String link = null;
 489     try {
 490       link = getElasticFullUrl("/" + sse.getId(), getIndexName());
 491     } catch (Exception exc) {
 492       LOG.error(AaiUiMsgs.ES_LINK_UPSERT, exc.getLocalizedMessage());
 493       return;
 494     }
 495
 496     boolean wasEntryDiscovered = false;
 497     if (esGetTxn.getOperationResult().getResultCode() == 404) {
 498       LOG.info(AaiUiMsgs.ES_SIMPLE_PUT, sse.getEntityPrimaryKeyValue());
 499     } else if (esGetTxn.getOperationResult().getResultCode() == 200) {
 500       wasEntryDiscovered = true;
 501     } else {
 502       /*
 503        * Not being a 200 does not mean a failure. eg 201 is returned for created. and 500 for es not
 504        * found TODO -> Should we return.
 505        */
 506       LOG.error(AaiUiMsgs.ES_OPERATION_RETURN_CODE,
 507           String.valueOf(esGetTxn.getOperationResult().getResultCode()));
 508       return;
 509     }
 510     // Insert a new document only if the paylod is different.
 511     // This is determined by hashing the payload and using it as a id for the document
 512     //
 513     if (!wasEntryDiscovered) {
 514       try {
 515         String jsonPayload = null;
 516
 517         jsonPayload = sse.getIndexDocumentJson();
 518         if (link != null && jsonPayload != null) {
 519
 520           NetworkTransaction updateElasticTxn = new NetworkTransaction();
 521           updateElasticTxn.setLink(link);
 522           updateElasticTxn.setEntityType(esGetTxn.getEntityType());
 523           updateElasticTxn.setDescriptor(esGetTxn.getDescriptor());
 524           updateElasticTxn.setOperationType(HttpMethod.PUT);
 525
 526           esWorkOnHand.incrementAndGet();
 527           supplyAsync(new PerformElasticSearchPut(jsonPayload, updateElasticTxn, esDataProvider),
 528               esPutExecutor).whenComplete((result, error) -> {
 529
 530                 esWorkOnHand.decrementAndGet();
 531
 532                 if (error != null) {
 533                   String message = "Suggestion search entity sync UPDATE PUT error - "
 534                       + error.getLocalizedMessage();
 535                   LOG.error(AaiUiMsgs.ES_SUGGESTION_SEARCH_ENTITY_SYNC_ERROR, message);
 536                 } else {
 537                   updateElasticSearchCounters(result);
 538                   processStoreDocumentResult(result, esGetTxn, sse);
 539                 }
 540               });
 541         }
 542       } catch (Exception exc) {
 543         String message =
 544             "Exception caught during suggestion search entity sync PUT operation. Message - "
 545                 + exc.getLocalizedMessage();
 546         LOG.error(AaiUiMsgs.ES_SUGGESTION_SEARCH_ENTITY_SYNC_ERROR, message);
 547       }
 548     }
 549   }
 550
 551   private void processStoreDocumentResult(NetworkTransaction esPutResult,
 552       NetworkTransaction esGetResult, SuggestionSearchEntity sse) {
 553
 554     OperationResult or = esPutResult.getOperationResult();
 555
 556     if (!or.wasSuccessful()) {
 557       if (or.getResultCode() == VERSION_CONFLICT_EXCEPTION_CODE) {
 558
 559         if (shouldAllowRetry(sse.getId())) {
 560           esWorkOnHand.incrementAndGet();
 561
 562           RetrySuggestionEntitySyncContainer rssec =
 563               new RetrySuggestionEntitySyncContainer(esGetResult, sse);
 564           retryQueue.push(rssec);
 565
 566           String message = "Store document failed during suggestion search entity synchronization"
 567               + " due to version conflict. Entity will be re-synced.";
 568           LOG.warn(AaiUiMsgs.ES_SUGGESTION_SEARCH_ENTITY_SYNC_ERROR, message);
 569         }
 570       } else {
 571         String message =
 572             "Store document failed during suggestion search entity synchronization with result code "
 573                 + or.getResultCode() + " and result message " + or.getResult();
 574         LOG.error(AaiUiMsgs.ES_SUGGESTION_SEARCH_ENTITY_SYNC_ERROR, message);
 575       }
 576     }
 577   }
 578
 579   /**
 580    * Perform retry sync.
 581    */
 582   private void performRetrySync() {
 583     while (retryQueue.peek() != null) {
 584
 585       RetrySuggestionEntitySyncContainer susc = retryQueue.poll();
 586       if (susc != null) {
 587
 588         SuggestionSearchEntity sus = susc.getSuggestionSearchEntity();
 589         NetworkTransaction txn = susc.getNetworkTransaction();
 590
 591         String link = null;
 592         try {
 593           /*
 594            * In this retry flow the se object has already derived its fields
 595            */
 596           link = getElasticFullUrl("/" + sus.getId(), getIndexName());
 597         } catch (Exception exc) {
 598           LOG.error(AaiUiMsgs.ES_FAILED_TO_CONSTRUCT_URI, exc.getLocalizedMessage());
 599         }
 600
 601         if (link != null) {
 602           NetworkTransaction retryTransaction = new NetworkTransaction();
 603           retryTransaction.setLink(link);
 604           retryTransaction.setEntityType(txn.getEntityType());
 605           retryTransaction.setDescriptor(txn.getDescriptor());
 606           retryTransaction.setOperationType(HttpMethod.GET);
 607
 608           /*
 609            * IMPORTANT - DO NOT incrementAndGet the esWorkOnHand as this is a retry flow! We already
 610            * called incrementAndGet when queuing the failed PUT!
 611            */
 612
 613           supplyAsync(new PerformElasticSearchRetrieval(retryTransaction, esDataProvider),
 614               esExecutor).whenComplete((result, error) -> {
 615
 616                 esWorkOnHand.decrementAndGet();
 617
 618                 if (error != null) {
 619                   LOG.error(AaiUiMsgs.ES_RETRIEVAL_FAILED_RESYNC, error.getLocalizedMessage());
 620                 } else {
 621                   updateElasticSearchCounters(result);
 622                   performDocumentUpsert(result, sus);
 623                 }
 624               });
 625         }
 626
 627       }
 628     }
 629   }
 630
 631   /**
 632    * Should allow retry.
 633    *
 634    * @param id the id
 635    * @return true, if successful
 636    */
 637   private boolean shouldAllowRetry(String id) {
 638     boolean isRetryAllowed = true;
 639     if (retryLimitTracker.get(id) != null) {
 640       Integer currentCount = retryLimitTracker.get(id);
 641       if (currentCount.intValue() >= RETRY_COUNT_PER_ENTITY_LIMIT.intValue()) {
 642         isRetryAllowed = false;
 643         String message = "Searchable entity re-sync limit reached for " + id
 644             + ", re-sync will no longer be attempted for this entity";
 645         LOG.error(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 646       } else {
 647         Integer newCount = new Integer(currentCount.intValue() + 1);
 648         retryLimitTracker.put(id, newCount);
 649       }
 650     } else {
 651       Integer firstRetryCount = new Integer(1);
 652       retryLimitTracker.put(id, firstRetryCount);
 653     }
 654
 655     return isRetryAllowed;
 656   }
 657
 658
 659
 660   @Override
 661   public SynchronizerState getState() {
 662
 663     if (!isSyncDone()) {
 664       return SynchronizerState.PERFORMING_SYNCHRONIZATION;
 665     }
 666
 667     return SynchronizerState.IDLE;
 668
 669   }
 670
 671   /*
 672    * (non-Javadoc)
 673    *
 674    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#getStatReport(boolean)
 675    */
 676   @Override
 677   public String getStatReport(boolean showFinalReport) {
 678           syncDurationInMs = System.currentTimeMillis() - syncStartedTimeStampInMs;
 679           return getStatReport(syncDurationInMs, showFinalReport);
 680   }
 681
 682   /*
 683    * (non-Javadoc)
 684    *
 685    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#shutdown()
 686    */
 687   @Override
 688   public void shutdown() {
 689     this.shutdownExecutors();
 690   }
 691
 692   @Override
 693   protected boolean isSyncDone() {
 694
 695     int totalWorkOnHand = aaiWorkOnHand.get() + esWorkOnHand.get();
 696
 697     if (LOG.isDebugEnabled()) {
 698       LOG.debug(AaiUiMsgs.DEBUG_GENERIC, indexName + ", isSyncDone(), totalWorkOnHand = "
 699           + totalWorkOnHand + " all work enumerated = " + allWorkEnumerated);
 700     }
 701
 702     if (totalWorkOnHand > 0 || !allWorkEnumerated) {
 703       return false;
 704     }
 705
 706     this.syncInProgress = false;
 707
 708     return true;
 709   }
 710
 711   /*
 712    * (non-Javadoc)
 713    *
 714    * @see org.openecomp.sparky.synchronizer.AbstractEntitySynchronizer#clearCache()
 715    */
 716   @Override
 717   public void clearCache() {
 718
 719     if (syncInProgress) {
 720       LOG.debug(AaiUiMsgs.DEBUG_GENERIC,
 721           "Autosuggestion Entity Summarizer in progress, request to clear cache ignored");
 722       return;
 723     }
 724
 725     super.clearCache();
 726     this.resetCounters();
 727     if (entityCounters != null) {
 728       entityCounters.clear();
 729     }
 730
 731     allWorkEnumerated = false;
 732
 733   }
 734
 735 }