Merge pull request #98 from apriltuesday/EVA-4056

EVA-4056: Add trace logging and release unused accessions when prefiltering

Author: apriltuesday

accession-commons-core/src/main/java/uk/ac/ebi/ampt2d/commons/accession/core/BasicAccessioningService.java

@@ -33,6 +33,7 @@
 
 import java.io.Serializable;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -94,6 +95,8 @@ private Map<HASH, MODEL> mapHashOfMessages(List<? extends MODEL> messages) {
      * See {@link #getPreexistingAccessions(List)} } for more details.
      */
     private List<GetOrCreateAccessionWrapper<MODEL, HASH, ACCESSION>> saveAccessions(List<AccessionWrapper<MODEL, HASH, ACCESSION>> accessions) {
+        logger.trace("Accessions to save: {}", accessions.stream().map(AccessionWrapper::getAccession).collect(
+                Collectors.toList()));
         switch (this.accessionSaveMode) {
             case PREFILTER_EXISTING:
                 return saveAccessionsPrefilteringExisting(accessions);
@@ -130,6 +133,16 @@ private List<GetOrCreateAccessionWrapper<MODEL, HASH, ACCESSION>> saveAccessions
         List<AccessionWrapper<MODEL, HASH, ACCESSION>> preexistingAccessions = dbService.findAllByHash(allHashes);
         Set<HASH> preexistingHashes = preexistingAccessions.stream().map(AccessionWrapper::getHash).collect(Collectors.toSet());
 
+        // release accessions associated with pre-existing hashes
+        Set<ACCESSION> accessionsToRelease = accessions
+                .stream()
+                .filter(accession -> preexistingHashes.contains(accession.getHash()))
+                .map(AccessionWrapper::getAccession)
+                .collect(Collectors.toSet());
+        SaveResponse<ACCESSION> response = new SaveResponse<>(Collections.emptySet(), accessionsToRelease);
+        accessionGenerator.postSave(response);
+
+        // save rest of the accessions
         List<AccessionWrapper<MODEL, HASH, ACCESSION>> accessionsToSave = accessions.stream()
                 .filter(accession -> !preexistingHashes.contains(accession.getHash()))
                 .collect(Collectors.toList());

accession-commons-monotonic-generator-jpa/src/main/java/uk/ac/ebi/ampt2d/commons/accession/generators/monotonic/BlockManager.java

@@ -17,6 +17,8 @@
  */
 package uk.ac.ebi.ampt2d.commons.accession.generators.monotonic;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.data.util.Pair;
 import uk.ac.ebi.ampt2d.commons.accession.core.exceptions.AccessionCouldNotBeGeneratedException;
 import uk.ac.ebi.ampt2d.commons.accession.core.exceptions.AccessionIsNotPendingException;
@@ -36,6 +38,8 @@
  */
 class BlockManager {
 
+    private static final Logger logger = LoggerFactory.getLogger(BlockManager.class);
+
     private final PriorityQueue<ContiguousIdBlock> assignedBlocks;
 
     private final MonotonicRangePriorityQueue availableRanges;
@@ -52,6 +56,7 @@ public BlockManager() {
     }
 
     public void addBlock(ContiguousIdBlock block) {
+        logger.trace("Adding block: {}", block);
         assignedBlocks.add(block);
         availableRanges.add(new MonotonicRange(block.getLastCommitted() + 1, block.getLastValue()));
     }
@@ -67,12 +72,14 @@ public MonotonicRangePriorityQueue getAvailableRanges() {
      * @return Array of monotonically increasing IDs
      */
     public long[] pollNext(int maxValues) throws AccessionCouldNotBeGeneratedException {
+        logger.trace("Polling for {} values", maxValues);
         if (!hasAvailableAccessions(maxValues)) {
             throw new AccessionCouldNotBeGeneratedException("Block manager doesn't have " + maxValues + " values available.");
         }
         MonotonicRange monotonicRange = pollNextMonotonicRange(maxValues);
         long[] ids = monotonicRange.getIds();
         generatedAccessions.addAll(LongStream.of(ids).boxed().collect(Collectors.toList()));
+        logger.trace("Generated accessions: {}", ids);
         return ids;
     }
 
@@ -98,6 +105,7 @@ public boolean hasAvailableAccessions(int accessionsNeeded) {
     }
 
     public Set<ContiguousIdBlock> commit(long[] accessions) throws AccessionIsNotPendingException {
+        logger.trace("Inside commit for accessions: {}", accessions);
         assertAccessionsArePending(accessions);
         return doCommit(accessions);
     }
@@ -119,17 +127,31 @@ private Set<ContiguousIdBlock> doCommit(long[] accessions) {
         addToCommitted(accessions);
 
         ContiguousIdBlock block = assignedBlocks.peek();
-        while (block != null && committedAccessions.peek() != null &&
-                committedAccessions.peek() == block.getLastCommitted() + 1) {
-            //Next value continues sequence, change last committed value
+        logger.trace("Trying to commit within block: {}", block);
+        while (true) {
+            if (block == null) {
+                logger.trace("No more blocks");
+                break;
+            } else if (committedAccessions.peek() == null) {
+                logger.trace("No more accessions to commit");
+                break;
+            } else if (committedAccessions.peek() != block.getLastCommitted() + 1) {
+                logger.trace("Next accession to commit is not in sequence: {} != {} + 1",
+                             committedAccessions.peek(), block.getLastCommitted());
+                break;
+            }
+            // Next value continues sequence, change last committed value
+            logger.trace("Setting last committed to {}", committedAccessions.peek());
             block.setLastCommitted(committedAccessions.poll());
             blocksToUpdate.add(block);
             if (!block.isNotFull()) {
                 assignedBlocks.poll();
                 block = assignedBlocks.peek();
+                logger.trace("Trying to commit within block: {}", block);
             }
         }
 
+        logger.trace("Blocks to update: {}", blocksToUpdate);
         return blocksToUpdate;
     }
 
@@ -141,6 +163,7 @@ private void addToCommitted(long[] accessions) {
     }
 
     public void release(long[] accessions) throws AccessionIsNotPendingException {
+        logger.trace("Inside release for accessions: {}", accessions);
         assertAccessionsArePending(accessions);
         doRelease(accessions);
     }
@@ -158,6 +181,7 @@ private void doRelease(long[] accessions) {
      * @throws AccessionIsNotPendingException When the generated accession does not match with the accession to commit
      */
     public Set<ContiguousIdBlock> recoverState(long[] committedElements) throws AccessionIsNotPendingException {
+        logger.trace("Inside recoverState for accessions: {}", committedElements);
         List<MonotonicRange> ranges = MonotonicRange.convertToMonotonicRanges(committedElements);
         List<MonotonicRange> newAvailableRanges = new ArrayList<>();
         for (MonotonicRange monotonicRange : this.availableRanges) {

accession-commons-monotonic-generator-jpa/src/main/java/uk/ac/ebi/ampt2d/commons/accession/generators/monotonic/MonotonicAccessionGenerator.java

@@ -17,6 +17,8 @@
  */
 package uk.ac.ebi.ampt2d.commons.accession.generators.monotonic;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import uk.ac.ebi.ampt2d.commons.accession.block.initialization.BlockInitializationException;
 import uk.ac.ebi.ampt2d.commons.accession.core.exceptions.AccessionCouldNotBeGeneratedException;
 import uk.ac.ebi.ampt2d.commons.accession.core.exceptions.AccessionGeneratorShutDownException;
@@ -43,6 +45,8 @@
  */
 public class MonotonicAccessionGenerator<MODEL> implements AccessionGenerator<MODEL, Long> {
 
+    private static final Logger logger = LoggerFactory.getLogger(MonotonicAccessionGenerator.class);
+
     private final BlockManager blockManager;
     private final String categoryId;
     private final ContiguousIdBlockService blockService;
@@ -68,6 +72,7 @@ private static void assertBlockParametersAreInitialized(ContiguousIdBlockService
     }
 
     private boolean recoverAndReserveUncompletedBlock(String applicationInstanceId) {
+        logger.trace("Reserving incomplete block");
         if (monotonicDatabaseService != null) {
             ContiguousIdBlock uncompletedBlock = blockService
                     .reserveFirstUncompletedBlockForCategoryIdAndApplicationInstanceId(categoryId, applicationInstanceId);
@@ -100,6 +105,7 @@ private void recoverStateForElements(long[] committedElements) throws AccessionI
     public synchronized long[] generateAccessions(int numAccessionsToGenerate, String applicationInstanceId)
             throws AccessionCouldNotBeGeneratedException {
         checkAccessionGeneratorNotShutDown();
+        logger.trace("Generating {} accessions for application ID {}", numAccessionsToGenerate, applicationInstanceId);
         long[] accessions = new long[numAccessionsToGenerate];
         reserveBlocksUntilSizeIs(numAccessionsToGenerate, applicationInstanceId);
 
@@ -128,6 +134,7 @@ private synchronized void reserveBlocksUntilSizeIs(int totalAccessionsToGenerate
     }
 
     private synchronized void reserveBlock(String categoryId, String instanceId) {
+        logger.trace("Inside reserveBlock");
         if (UNCOMPLETED_BLOCKS_AVAILABLE) {
             boolean reservedUncompleted = recoverAndReserveUncompletedBlock(instanceId);
             if (!reservedUncompleted) {
@@ -140,6 +147,7 @@ private synchronized void reserveBlock(String categoryId, String instanceId) {
 
 
     private synchronized void reserveNewBlock(String categoryId, String instanceId) {
+        logger.trace("Reserving new block");
         blockManager.addBlock(blockService.reserveNewBlock(categoryId, instanceId));
     }
 
@@ -183,7 +191,7 @@ public synchronized void postSave(SaveResponse<Long> response) {
     public void shutDownAccessionGenerator() {
         List<ContiguousIdBlock> blockList = blockManager.getAssignedBlocks();
         blockList.stream().forEach(block -> block.releaseReserved());
-        blockService.save(blockList);
+        ExponentialBackOff.execute(() -> blockService.save(blockList), 10, 30);
         blockManager.shutDownBlockManager();
         SHUTDOWN = true;
     }