src/main/java/org/onap/aai/sparky/viewinspect/sync/ViewInspectEntitySynchronizer.java

   1 /**
   2  * ============LICENSE_START=======================================================
   3  * org.onap.aai
   4  * ================================================================================
   5  * Copyright © 2017 AT&T Intellectual Property. All rights reserved.
   6  * Copyright © 2017 Amdocs
   7  * ================================================================================
   8  * Licensed under the Apache License, Version 2.0 (the "License");
   9  * you may not use this file except in compliance with the License.
  10  * You may obtain a copy of the License at
  11  *
  12  *       http://www.apache.org/licenses/LICENSE-2.0
  13  *
  14  * Unless required by applicable law or agreed to in writing, software
  15  * distributed under the License is distributed on an "AS IS" BASIS,
  16  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  17  * See the License for the specific language governing permissions and
  18  * limitations under the License.
  19  * ============LICENSE_END=========================================================
  20  *
  21  * ECOMP is a trademark and service mark of AT&T Intellectual Property.
  22  */
  23 package org.onap.aai.sparky.viewinspect.sync;
  24
  25 import static java.util.concurrent.CompletableFuture.supplyAsync;
  26
  27 import java.io.IOException;
  28 import java.util.ArrayList;
  29 import java.util.Collection;
  30 import java.util.Deque;
  31 import java.util.Iterator;
  32 import java.util.List;
  33 import java.util.Map;
  34 import java.util.concurrent.ConcurrentHashMap;
  35 import java.util.concurrent.ConcurrentLinkedDeque;
  36 import java.util.concurrent.ExecutorService;
  37 import java.util.function.Supplier;
  38
  39 import org.onap.aai.cl.api.Logger;
  40 import org.onap.aai.cl.eelf.LoggerFactory;
  41 import org.onap.aai.cl.mdc.MdcContext;
  42 import org.onap.aai.restclient.client.OperationResult;
  43 import org.onap.aai.sparky.config.oxm.OxmEntityDescriptor;
  44 import org.onap.aai.sparky.config.oxm.OxmEntityLookup;
  45 import org.onap.aai.sparky.config.oxm.SearchableEntityLookup;
  46 import org.onap.aai.sparky.config.oxm.SearchableOxmEntityDescriptor;
  47 import org.onap.aai.sparky.dal.ActiveInventoryAdapter;
  48 import org.onap.aai.sparky.dal.NetworkTransaction;
  49 import org.onap.aai.sparky.dal.rest.HttpMethod;
  50 import org.onap.aai.sparky.logging.AaiUiMsgs;
  51 import org.onap.aai.sparky.sync.AbstractEntitySynchronizer;
  52 import org.onap.aai.sparky.sync.IndexSynchronizer;
  53 import org.onap.aai.sparky.sync.SynchronizerConstants;
  54 import org.onap.aai.sparky.sync.config.ElasticSearchSchemaConfig;
  55 import org.onap.aai.sparky.sync.config.NetworkStatisticsConfig;
  56 import org.onap.aai.sparky.sync.entity.MergableEntity;
  57 import org.onap.aai.sparky.sync.entity.SearchableEntity;
  58 import org.onap.aai.sparky.sync.entity.SelfLinkDescriptor;
  59 import org.onap.aai.sparky.sync.enumeration.OperationState;
  60 import org.onap.aai.sparky.sync.enumeration.SynchronizerState;
  61 import org.onap.aai.sparky.sync.task.PerformActiveInventoryRetrieval;
  62 import org.onap.aai.sparky.sync.task.PerformElasticSearchPut;
  63 import org.onap.aai.sparky.sync.task.PerformElasticSearchRetrieval;
  64 import org.onap.aai.sparky.sync.task.PerformElasticSearchUpdate;
  65 import org.onap.aai.sparky.util.NodeUtils;
  66 import org.slf4j.MDC;
  67
  68 import com.fasterxml.jackson.core.JsonProcessingException;
  69 import com.fasterxml.jackson.databind.JsonNode;
  70 import com.fasterxml.jackson.databind.ObjectReader;
  71 import com.fasterxml.jackson.databind.node.ArrayNode;
  72
  73 /**
  74  * The Class SearchableEntitySynchronizer.
  75  */
  76 public class ViewInspectEntitySynchronizer extends AbstractEntitySynchronizer
  77     implements IndexSynchronizer {
  78
  79   /**
  80    * The Class RetrySearchableEntitySyncContainer.
  81    */
  82   private class RetrySearchableEntitySyncContainer {
  83     NetworkTransaction txn;
  84     SearchableEntity se;
  85
  86     /**
  87      * Instantiates a new retry searchable entity sync container.
  88      *
  89      * @param txn the txn
  90      * @param se the se
  91      */
  92     public RetrySearchableEntitySyncContainer(NetworkTransaction txn, SearchableEntity se) {
  93       this.txn = txn;
  94       this.se = se;
  95     }
  96
  97     public NetworkTransaction getNetworkTransaction() {
  98       return txn;
  99     }
 100
 101     public SearchableEntity getSearchableEntity() {
 102       return se;
 103     }
 104   }
 105
 106   private static final Logger LOG =
 107       LoggerFactory.getInstance().getLogger(ViewInspectEntitySynchronizer.class);
 108
 109   private boolean allWorkEnumerated;
 110   private Deque<SelfLinkDescriptor> selflinks;
 111   private Deque<RetrySearchableEntitySyncContainer> retryQueue;
 112   private Map<String, Integer> retryLimitTracker;
 113   protected ExecutorService esPutExecutor;
 114   private OxmEntityLookup oxmEntityLookup;
 115   private SearchableEntityLookup searchableEntityLookup;
 116
 117   /**
 118    * Instantiates a new searchable entity synchronizer.
 119    *
 120    * @param indexName the index name
 121    * @throws Exception the exception
 122    */
 123   public ViewInspectEntitySynchronizer(ElasticSearchSchemaConfig schemaConfig,
 124       int internalSyncWorkers, int aaiWorkers, int esWorkers, NetworkStatisticsConfig aaiStatConfig,
 125       NetworkStatisticsConfig esStatConfig, OxmEntityLookup oxmEntityLookup,
 126       SearchableEntityLookup searchableEntityLookup) throws Exception {
 127     super(LOG, "SES", internalSyncWorkers, aaiWorkers, esWorkers, schemaConfig.getIndexName(),
 128         aaiStatConfig, esStatConfig);
 129
 130     this.oxmEntityLookup = oxmEntityLookup;
 131     this.searchableEntityLookup = searchableEntityLookup;
 132     this.allWorkEnumerated = false;
 133     this.selflinks = new ConcurrentLinkedDeque<SelfLinkDescriptor>();
 134     this.retryQueue = new ConcurrentLinkedDeque<RetrySearchableEntitySyncContainer>();
 135     this.retryLimitTracker = new ConcurrentHashMap<String, Integer>();
 136     this.synchronizerName = "Searchable Entity Synchronizer";
 137     this.esPutExecutor = NodeUtils.createNamedExecutor("SES-ES-PUT", 5, LOG);
 138     this.aaiEntityStats.intializeEntityCounters(
 139         searchableEntityLookup.getSearchableEntityDescriptors().keySet());
 140     this.esEntityStats.intializeEntityCounters(
 141         searchableEntityLookup.getSearchableEntityDescriptors().keySet());
 142     this.syncDurationInMs = -1;
 143   }
 144
 145   /**
 146    * Collect all the work.
 147    *
 148    * @return the operation state
 149    */
 150   private OperationState collectAllTheWork() {
 151     final Map<String, String> contextMap = MDC.getCopyOfContextMap();
 152     Map<String, SearchableOxmEntityDescriptor> descriptorMap =
 153         searchableEntityLookup.getSearchableEntityDescriptors();
 154
 155     if (descriptorMap.isEmpty()) {
 156       LOG.error(AaiUiMsgs.ERROR_LOADING_OXM_SEARCHABLE_ENTITIES);
 157       LOG.info(AaiUiMsgs.ERROR_LOADING_OXM_SEARCHABLE_ENTITIES);
 158       return OperationState.ERROR;
 159     }
 160
 161     Collection<String> syncTypes = descriptorMap.keySet();
 162
 163     /*Collection<String> syncTypes = new ArrayList<String>();
 164     syncTypes.add("service-instance");*/
 165
 166     try {
 167
 168       /*
 169        * launch a parallel async thread to process the documents for each entity-type (to max the
 170        * of the configured executor anyway)
 171        */
 172
 173       aaiWorkOnHand.set(syncTypes.size());
 174
 175       for (String key : syncTypes) {
 176
 177         supplyAsync(new Supplier<Void>() {
 178
 179           @Override
 180           public Void get() {
 181             MDC.setContextMap(contextMap);
 182             OperationResult typeLinksResult = null;
 183             try {
 184               typeLinksResult = aaiAdapter.getSelfLinksByEntityType(key);
 185               aaiWorkOnHand.decrementAndGet();
 186               processEntityTypeSelfLinks(typeLinksResult);
 187             } catch (Exception exc) {
 188               // TODO -> LOG, what should be logged here?
 189             }
 190
 191             return null;
 192           }
 193
 194         }, aaiExecutor).whenComplete((result, error) -> {
 195
 196           if (error != null) {
 197             LOG.error(AaiUiMsgs.ERROR_GENERIC,
 198                 "An error occurred getting data from AAI. Error = " + error.getMessage());
 199           }
 200         });
 201
 202       }
 203
 204       while (aaiWorkOnHand.get() != 0) {
 205
 206         if (LOG.isDebugEnabled()) {
 207           LOG.debug(AaiUiMsgs.WAIT_FOR_ALL_SELFLINKS_TO_BE_COLLECTED);
 208         }
 209
 210         Thread.sleep(1000);
 211       }
 212
 213       aaiWorkOnHand.set(selflinks.size());
 214       allWorkEnumerated = true;
 215       syncEntityTypes();
 216
 217       while (!isSyncDone()) {
 218         performRetrySync();
 219         Thread.sleep(1000);
 220       }
 221
 222       /*
 223        * Make sure we don't hang on to retries that failed which could cause issues during future
 224        * syncs
 225        */
 226       retryLimitTracker.clear();
 227
 228     } catch (Exception exc) {
 229       // TODO -> LOG, waht should be logged here?
 230     }
 231
 232     return OperationState.OK;
 233   }
 234
 235   /* (non-Javadoc)
 236    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#doSync()
 237    */
 238   @Override
 239   public OperationState doSync() {
 240     this.syncDurationInMs = -1;
 241     String txnID = NodeUtils.getRandomTxnId();
 242     MdcContext.initialize(txnID, "SearchableEntitySynchronizer", "", "Sync", "");
 243
 244     resetCounters();
 245     this.allWorkEnumerated = false;
 246     syncStartedTimeStampInMs = System.currentTimeMillis();
 247     collectAllTheWork();
 248
 249     return OperationState.OK;
 250   }
 251
 252   /**
 253    * Process entity type self links.
 254    *
 255    * @param operationResult the operation result
 256    */
 257   private void processEntityTypeSelfLinks(OperationResult operationResult) {
 258
 259     JsonNode rootNode = null;
 260
 261     final String jsonResult = operationResult.getResult();
 262
 263     if (jsonResult != null && jsonResult.length() > 0 && operationResult.wasSuccessful()) {
 264
 265       try {
 266         rootNode = mapper.readTree(jsonResult);
 267       } catch (IOException exc) {
 268         String message =
 269             "Could not deserialize JSON (representing operation result) as node tree. " +
 270             "Operation result = " + jsonResult + ". " + exc.getLocalizedMessage();
 271         LOG.error(AaiUiMsgs.JSON_PROCESSING_ERROR, message);
 272       }
 273
 274       JsonNode resultData = rootNode.get("result-data");
 275       ArrayNode resultDataArrayNode = null;
 276
 277       if (resultData.isArray()) {
 278         resultDataArrayNode = (ArrayNode) resultData;
 279
 280         Iterator<JsonNode> elementIterator = resultDataArrayNode.elements();
 281         JsonNode element = null;
 282
 283         while (elementIterator.hasNext()) {
 284           element = elementIterator.next();
 285
 286           final String resourceType = NodeUtils.getNodeFieldAsText(element, "resource-type");
 287           final String resourceLink = NodeUtils.getNodeFieldAsText(element, "resource-link");
 288
 289           SearchableOxmEntityDescriptor descriptor = null;
 290
 291           if (resourceType != null && resourceLink != null) {
 292
 293             descriptor = searchableEntityLookup.getSearchableEntityDescriptors().get(resourceType);
 294
 295             if (descriptor == null) {
 296               LOG.error(AaiUiMsgs.MISSING_ENTITY_DESCRIPTOR, resourceType);
 297               // go to next element in iterator
 298               continue;
 299             }
 300
 301             if (descriptor.hasSearchableAttributes()) {
 302               selflinks.add(new SelfLinkDescriptor(resourceLink, SynchronizerConstants.NODES_ONLY_MODIFIER, resourceType));
 303             }
 304
 305           }
 306         }
 307       }
 308     }
 309
 310   }
 311
 312   /**
 313    * Sync entity types.
 314    */
 315   private void syncEntityTypes() {
 316
 317     while (selflinks.peek() != null) {
 318
 319       SelfLinkDescriptor linkDescriptor = selflinks.poll();
 320       aaiWorkOnHand.decrementAndGet();
 321
 322       OxmEntityDescriptor descriptor = null;
 323
 324       if (linkDescriptor.getSelfLink() != null && linkDescriptor.getEntityType() != null) {
 325
 326         descriptor = oxmEntityLookup.getEntityDescriptors().get(linkDescriptor.getEntityType());
 327
 328         if (descriptor == null) {
 329           LOG.error(AaiUiMsgs.MISSING_ENTITY_DESCRIPTOR, linkDescriptor.getEntityType());
 330           // go to next element in iterator
 331           continue;
 332         }
 333
 334         NetworkTransaction txn = new NetworkTransaction();
 335         txn.setDescriptor(descriptor);
 336         txn.setLink(linkDescriptor.getSelfLink());
 337         txn.setOperationType(HttpMethod.GET);
 338         txn.setEntityType(linkDescriptor.getEntityType());
 339
 340         aaiWorkOnHand.incrementAndGet();
 341
 342         supplyAsync(new PerformActiveInventoryRetrieval(txn, aaiAdapter), aaiExecutor)
 343             .whenComplete((result, error) -> {
 344
 345               aaiWorkOnHand.decrementAndGet();
 346
 347               if (error != null) {
 348                 LOG.error(AaiUiMsgs.AAI_RETRIEVAL_FAILED_GENERIC, error.getLocalizedMessage());
 349               } else {
 350                 if (result == null) {
 351                   LOG.error(AaiUiMsgs.AAI_RETRIEVAL_FAILED_FOR_SELF_LINK,
 352                       linkDescriptor.getSelfLink());
 353                 } else {
 354                   updateActiveInventoryCounters(result);
 355                   fetchDocumentForUpsert(result);
 356                 }
 357               }
 358             });
 359       }
 360
 361     }
 362
 363   }
 364
 365   /**
 366    * Perform document upsert.
 367    *
 368    * @param esGetTxn the es get txn
 369    * @param se the se
 370    */
 371   protected void performDocumentUpsert(NetworkTransaction esGetTxn, SearchableEntity se) {
 372     /**
 373      * <p>
 374      * <ul>
 375      * As part of the response processing we need to do the following:
 376      * <li>1. Extract the version (if present), it will be the ETAG when we use the
 377      * Search-Abstraction-Service
 378      * <li>2. Spawn next task which is to do the PUT operation into elastic with or with the version
 379      * tag
 380      * <li>a) if version is null or RC=404, then standard put, no _update with version tag
 381      * <li>b) if version != null, do PUT with _update?version= versionNumber in the URI to elastic
 382      * </ul>
 383      * </p>
 384      */
 385     String link = null;
 386     try {
 387       link = elasticSearchAdapter.buildElasticSearchGetDocUrl(getIndexName(), se.getId());
 388     } catch (Exception exc) {
 389       LOG.error(AaiUiMsgs.ES_LINK_UPSERT, exc.getLocalizedMessage());
 390       return;
 391     }
 392
 393     String versionNumber = null;
 394     boolean wasEntryDiscovered = false;
 395     if (esGetTxn.getOperationResult().getResultCode() == 404) {
 396       LOG.info(AaiUiMsgs.ES_SIMPLE_PUT, se.getEntityPrimaryKeyValue());
 397     } else if (esGetTxn.getOperationResult().getResultCode() == 200) {
 398       wasEntryDiscovered = true;
 399       try {
 400         versionNumber = NodeUtils.extractFieldValueFromObject(
 401             NodeUtils.convertJsonStrToJsonNode(esGetTxn.getOperationResult().getResult()),
 402             "_version");
 403       } catch (IOException exc) {
 404         String message =
 405             "Error extracting version number from response, aborting searchable entity sync of "
 406                 + se.getEntityPrimaryKeyValue() + ". Error - " + exc.getLocalizedMessage();
 407         LOG.error(AaiUiMsgs.ERROR_EXTRACTING_FROM_RESPONSE, message);
 408         return;
 409       }
 410     } else {
 411       /*
 412        * Not being a 200 does not mean a failure. eg 201 is returned for created. TODO -> Should we
 413        * return.
 414        */
 415       LOG.error(AaiUiMsgs.ES_OPERATION_RETURN_CODE,
 416           String.valueOf(esGetTxn.getOperationResult().getResultCode()));
 417       return;
 418     }
 419
 420     try {
 421       String jsonPayload = null;
 422       if (wasEntryDiscovered) {
 423         try {
 424           ArrayList<JsonNode> sourceObject = new ArrayList<JsonNode>();
 425           NodeUtils.extractObjectsByKey(
 426               NodeUtils.convertJsonStrToJsonNode(esGetTxn.getOperationResult().getResult()),
 427               "_source", sourceObject);
 428
 429           if (!sourceObject.isEmpty()) {
 430             String responseSource = NodeUtils.convertObjectToJson(sourceObject.get(0), false);
 431             MergableEntity me = mapper.readValue(responseSource, MergableEntity.class);
 432             ObjectReader updater = mapper.readerForUpdating(me);
 433             MergableEntity merged = updater.readValue(NodeUtils.convertObjectToJson(se,false));
 434             jsonPayload = mapper.writeValueAsString(merged);
 435           }
 436         } catch (IOException exc) {
 437           String message =
 438               "Error extracting source value from response, aborting searchable entity sync of "
 439                   + se.getEntityPrimaryKeyValue() + ". Error - " + exc.getLocalizedMessage();
 440           LOG.error(AaiUiMsgs.ERROR_EXTRACTING_FROM_RESPONSE, message);
 441           return;
 442         }
 443       } else {
 444           jsonPayload = se.getAsJson();
 445       }
 446
 447       if (wasEntryDiscovered) {
 448         if (versionNumber != null && jsonPayload != null) {
 449
 450           String requestPayload = elasticSearchAdapter.buildBulkImportOperationRequest(getIndexName(),
 451               "default", se.getId(), versionNumber, jsonPayload);
 452
 453           NetworkTransaction transactionTracker = new NetworkTransaction();
 454           transactionTracker.setEntityType(esGetTxn.getEntityType());
 455           transactionTracker.setDescriptor(esGetTxn.getDescriptor());
 456           transactionTracker.setOperationType(HttpMethod.PUT);
 457
 458           esWorkOnHand.incrementAndGet();
 459           supplyAsync(new PerformElasticSearchUpdate(elasticSearchAdapter.getBulkUrl(),
 460               requestPayload, elasticSearchAdapter, transactionTracker), esPutExecutor)
 461                   .whenComplete((result, error) -> {
 462
 463                     esWorkOnHand.decrementAndGet();
 464
 465                     if (error != null) {
 466                       String message = "Searchable entity sync UPDATE PUT error - "
 467                           + error.getLocalizedMessage();
 468                       LOG.error(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 469                     } else {
 470                       updateElasticSearchCounters(result);
 471                       processStoreDocumentResult(result, esGetTxn, se);
 472                     }
 473                   });
 474         }
 475
 476       } else {
 477
 478         if (link != null && jsonPayload != null) {
 479
 480           NetworkTransaction updateElasticTxn = new NetworkTransaction();
 481           updateElasticTxn.setLink(link);
 482           updateElasticTxn.setEntityType(esGetTxn.getEntityType());
 483           updateElasticTxn.setDescriptor(esGetTxn.getDescriptor());
 484           updateElasticTxn.setOperationType(HttpMethod.PUT);
 485
 486           esWorkOnHand.incrementAndGet();
 487           supplyAsync(new PerformElasticSearchPut(jsonPayload, updateElasticTxn, elasticSearchAdapter),
 488               esPutExecutor).whenComplete((result, error) -> {
 489
 490                 esWorkOnHand.decrementAndGet();
 491
 492                 if (error != null) {
 493                   String message =
 494                       "Searchable entity sync UPDATE PUT error - " + error.getLocalizedMessage();
 495                   LOG.error(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 496                 } else {
 497                   updateElasticSearchCounters(result);
 498                   processStoreDocumentResult(result, esGetTxn, se);
 499                 }
 500               });
 501         }
 502       }
 503     } catch (Exception exc) {
 504       String message = "Exception caught during searchable entity sync PUT operation. Message - "
 505           + exc.getLocalizedMessage();
 506       LOG.error(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 507     }
 508   }
 509
 510   /**
 511    * Populate searchable entity document.
 512    *
 513    * @param doc the doc
 514    * @param result the result
 515    * @param resultDescriptor the result descriptor
 516    * @throws JsonProcessingException the json processing exception
 517    * @throws IOException Signals that an I/O exception has occurred.
 518    */
 519   protected void populateSearchableEntityDocument(SearchableEntity doc, String result,
 520       OxmEntityDescriptor resultDescriptor) throws JsonProcessingException, IOException {
 521
 522     doc.setEntityType(resultDescriptor.getEntityName());
 523
 524     JsonNode entityNode = mapper.readTree(result);
 525
 526     List<String> primaryKeyValues = new ArrayList<String>();
 527     String pkeyValue = null;
 528
 529     SearchableOxmEntityDescriptor searchableDescriptor = searchableEntityLookup.getSearchableEntityDescriptors().get(resultDescriptor.getEntityName());
 530
 531     for (String keyName : searchableDescriptor.getPrimaryKeyAttributeNames()) {
 532       pkeyValue = NodeUtils.getNodeFieldAsText(entityNode, keyName);
 533       if (pkeyValue != null) {
 534         primaryKeyValues.add(pkeyValue);
 535       } else {
 536         String message = "populateSearchableEntityDocument(), pKeyValue is null for entityType = "
 537             + resultDescriptor.getEntityName();
 538         LOG.warn(AaiUiMsgs.WARN_GENERIC, message);
 539       }
 540     }
 541
 542     final String primaryCompositeKeyValue = NodeUtils.concatArray(primaryKeyValues, "/");
 543     doc.setEntityPrimaryKeyValue(primaryCompositeKeyValue);
 544
 545     final List<String> searchTagFields = searchableDescriptor.getSearchableAttributes();
 546
 547     /*
 548      * Based on configuration, use the configured field names for this entity-Type to build a
 549      * multi-value collection of search tags for elastic search entity search criteria.
 550      */
 551     for (String searchTagField : searchTagFields) {
 552       String searchTagValue = NodeUtils.getNodeFieldAsText(entityNode, searchTagField);
 553       if (searchTagValue != null && !searchTagValue.isEmpty()) {
 554         doc.addSearchTagWithKey(searchTagValue, searchTagField);
 555       }
 556     }
 557   }
 558
 559   /**
 560    * Fetch document for upsert.
 561    *
 562    * @param txn the txn
 563    */
 564   private void fetchDocumentForUpsert(NetworkTransaction txn) {
 565     if (!txn.getOperationResult().wasSuccessful()) {
 566       String message = "Self link failure. Result - " + txn.getOperationResult().getResult();
 567       LOG.error(AaiUiMsgs.ERROR_GENERIC, message);
 568       return;
 569     }
 570
 571     SearchableOxmEntityDescriptor searchableDescriptor = searchableEntityLookup
 572         .getSearchableEntityDescriptors().get(txn.getDescriptor().getEntityName());
 573
 574     try {
 575       if (searchableDescriptor.hasSearchableAttributes()) {
 576
 577         final String jsonResult = txn.getOperationResult().getResult();
 578         if (jsonResult != null && jsonResult.length() > 0) {
 579
 580           SearchableEntity se = new SearchableEntity();
 581           se.setLink(ActiveInventoryAdapter.extractResourcePath(txn.getLink()));
 582           populateSearchableEntityDocument(se, jsonResult, txn.getDescriptor());
 583           se.deriveFields();
 584
 585           String link = null;
 586           try {
 587             link = elasticSearchAdapter.buildElasticSearchGetDocUrl(getIndexName(), se.getId());
 588           } catch (Exception exc) {
 589             LOG.error(AaiUiMsgs.ES_FAILED_TO_CONSTRUCT_QUERY, exc.getLocalizedMessage());
 590           }
 591
 592           if (link != null) {
 593             NetworkTransaction n2 = new NetworkTransaction();
 594             n2.setLink(link);
 595             n2.setEntityType(txn.getEntityType());
 596             n2.setDescriptor(txn.getDescriptor());
 597             n2.setOperationType(HttpMethod.GET);
 598
 599             esWorkOnHand.incrementAndGet();
 600
 601             supplyAsync(new PerformElasticSearchRetrieval(n2, elasticSearchAdapter), esExecutor)
 602                 .whenComplete((result, error) -> {
 603
 604                   esWorkOnHand.decrementAndGet();
 605
 606                   if (error != null) {
 607                     LOG.error(AaiUiMsgs.ES_RETRIEVAL_FAILED, error.getLocalizedMessage());
 608                   } else {
 609                     updateElasticSearchCounters(result);
 610                     performDocumentUpsert(result, se);
 611                   }
 612                 });
 613           }
 614         }
 615
 616       }
 617     } catch (JsonProcessingException exc) {
 618       // TODO -> LOG, waht should be logged here?
 619     } catch (IOException exc) {
 620       // TODO -> LOG, waht should be logged here?
 621     }
 622   }
 623
 624   /**
 625    * Process store document result.
 626    *
 627    * @param esPutResult the es put result
 628    * @param esGetResult the es get result
 629    * @param se the se
 630    */
 631   private void processStoreDocumentResult(NetworkTransaction esPutResult,
 632       NetworkTransaction esGetResult, SearchableEntity se) {
 633
 634     OperationResult or = esPutResult.getOperationResult();
 635
 636     if (!or.wasSuccessful()) {
 637       if (or.getResultCode() == VERSION_CONFLICT_EXCEPTION_CODE) {
 638
 639         if (shouldAllowRetry(se.getId())) {
 640           esWorkOnHand.incrementAndGet();
 641
 642           RetrySearchableEntitySyncContainer rsc =
 643               new RetrySearchableEntitySyncContainer(esGetResult, se);
 644           retryQueue.push(rsc);
 645
 646           String message = "Store document failed during searchable entity synchronization"
 647               + " due to version conflict. Entity will be re-synced.";
 648           LOG.warn(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 649         }
 650       } else {
 651         String message =
 652             "Store document failed during searchable entity synchronization with result code "
 653                 + or.getResultCode() + " and result message " + or.getResult();
 654         LOG.error(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 655       }
 656     }
 657   }
 658
 659   /**
 660    * Perform retry sync.
 661    */
 662   private void performRetrySync() {
 663     while (retryQueue.peek() != null) {
 664
 665       RetrySearchableEntitySyncContainer rsc = retryQueue.poll();
 666       if (rsc != null) {
 667
 668         SearchableEntity se = rsc.getSearchableEntity();
 669         NetworkTransaction txn = rsc.getNetworkTransaction();
 670
 671         String link = null;
 672         try {
 673           /*
 674            * In this retry flow the se object has already derived its fields
 675            */
 676           link = elasticSearchAdapter.buildElasticSearchGetDocUrl(getIndexName(), se.getId());
 677         } catch (Exception exc) {
 678           LOG.error(AaiUiMsgs.ES_FAILED_TO_CONSTRUCT_URI, exc.getLocalizedMessage());
 679         }
 680
 681         if (link != null) {
 682           NetworkTransaction retryTransaction = new NetworkTransaction();
 683           retryTransaction.setLink(link);
 684           retryTransaction.setEntityType(txn.getEntityType());
 685           retryTransaction.setDescriptor(txn.getDescriptor());
 686           retryTransaction.setOperationType(HttpMethod.GET);
 687
 688           /*
 689            * IMPORTANT - DO NOT incrementAndGet the esWorkOnHand as this is a retry flow! We already
 690            * called incrementAndGet when queuing the failed PUT!
 691            */
 692
 693           supplyAsync(new PerformElasticSearchRetrieval(retryTransaction, elasticSearchAdapter),
 694               esExecutor).whenComplete((result, error) -> {
 695
 696                 esWorkOnHand.decrementAndGet();
 697
 698                 if (error != null) {
 699                   LOG.error(AaiUiMsgs.ES_RETRIEVAL_FAILED_RESYNC, error.getLocalizedMessage());
 700                 } else {
 701                   updateElasticSearchCounters(result);
 702                   performDocumentUpsert(result, se);
 703                 }
 704               });
 705         }
 706
 707       }
 708     }
 709   }
 710
 711   /**
 712    * Should allow retry.
 713    *
 714    * @param id the id
 715    * @return true, if successful
 716    */
 717   private boolean shouldAllowRetry(String id) {
 718     boolean isRetryAllowed = true;
 719     if (retryLimitTracker.get(id) != null) {
 720       Integer currentCount = retryLimitTracker.get(id);
 721       if (currentCount.intValue() >= RETRY_COUNT_PER_ENTITY_LIMIT.intValue()) {
 722         isRetryAllowed = false;
 723         String message = "Searchable entity re-sync limit reached for " + id
 724             + ", re-sync will no longer be attempted for this entity";
 725         LOG.error(AaiUiMsgs.ES_SEARCHABLE_ENTITY_SYNC_ERROR, message);
 726       } else {
 727         Integer newCount = new Integer(currentCount.intValue() + 1);
 728         retryLimitTracker.put(id, newCount);
 729       }
 730     } else {
 731       Integer firstRetryCount = new Integer(1);
 732       retryLimitTracker.put(id, firstRetryCount);
 733     }
 734
 735     return isRetryAllowed;
 736   }
 737
 738   @Override
 739   public SynchronizerState getState() {
 740     if (!isSyncDone()) {
 741       return SynchronizerState.PERFORMING_SYNCHRONIZATION;
 742     }
 743
 744     return SynchronizerState.IDLE;
 745
 746   }
 747
 748   /* (non-Javadoc)
 749    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#getStatReport(boolean)
 750    */
 751   @Override
 752   public String getStatReport(boolean showFinalReport) {
 753     syncDurationInMs = System.currentTimeMillis() - syncStartedTimeStampInMs;
 754     return this.getStatReport(syncDurationInMs, showFinalReport);
 755   }
 756
 757   /* (non-Javadoc)
 758    * @see org.openecomp.sparky.synchronizer.IndexSynchronizer#shutdown()
 759    */
 760   @Override
 761   public void shutdown() {
 762     this.shutdownExecutors();
 763   }
 764
 765   @Override
 766   protected boolean isSyncDone() {
 767     int totalWorkOnHand = aaiWorkOnHand.get() + esWorkOnHand.get();
 768
 769     if (totalWorkOnHand > 0 || !allWorkEnumerated) {
 770       return false;
 771     }
 772
 773     return true;
 774   }
 775
 776 }