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