Add synchronized blocks around the deadLetters collection

To further protect the InMemorySequencedDeadLetterQueue against
concurrent changes on the dead-letters collection, we should add some
synchronization around it when accessing. Main point of attention here
are the evict and requeue operations which might remove a sequence or
bring it back. Due to this, contains and enqueue should also synchronize
The process operation does not receive any synchronized-blocks as it
works through the takenSequences set.

#2021

messaging/src/main/java/org/axonframework/messaging/deadletter/InMemorySequencedDeadLetterQueue.java

@@ -119,8 +119,10 @@ public void enqueue(@Nonnull Object sequenceIdentifier,
             }
         }
 
-        deadLetters.computeIfAbsent(toIdentifier(sequenceIdentifier), id -> new ConcurrentLinkedDeque<>())
-                   .addLast(letter);
+        synchronized (deadLetters) {
+            deadLetters.computeIfAbsent(toIdentifier(sequenceIdentifier), id -> new ConcurrentLinkedDeque<>())
+                       .addLast(letter);
+        }
     }
 
     @Override
@@ -132,14 +134,15 @@ public void evict(DeadLetter<? extends M> letter) {
                            .findFirst();
 
         if (optionalSequence.isPresent()) {
-            String sequenceId = optionalSequence.get().getKey();
-            Deque<DeadLetter<? extends M>> sequence = optionalSequence.get().getValue();
-            if (sequence.isEmpty()) {
-                logger.trace("Sequence [{}] is empty and will be removed.", sequenceId);
-                deadLetters.remove(sequenceId);
-            }
-            if (logger.isTraceEnabled()) {
-                logger.trace("Evicted letter [{}] for sequence [{}].", letter, sequenceId);
+            synchronized (deadLetters) {
+                String sequenceId = optionalSequence.get().getKey();
+                if (deadLetters.get(sequenceId).isEmpty()) {
+                    logger.trace("Sequence [{}] is empty and will be removed.", sequenceId);
+                    deadLetters.remove(sequenceId);
+                }
+                if (logger.isTraceEnabled()) {
+                    logger.trace("Evicted letter [{}] for sequence [{}].", letter, sequenceId);
+                }
             }
         } else if (logger.isDebugEnabled()) {
             logger.debug("Cannot evict [{}] as it could not be found in this queue.", letter);
@@ -158,11 +161,13 @@ public void requeue(
                            .findFirst();
 
         if (optionalSequence.isPresent()) {
-            String sequenceId = optionalSequence.get().getKey();
-            Deque<DeadLetter<? extends M>> sequence = optionalSequence.get().getValue();
-            sequence.addFirst(letterUpdater.apply(letter.markTouched()));
-            if (logger.isTraceEnabled()) {
-                logger.trace("Requeued letter [{}] for sequence [{}].", letter, sequenceId);
+            synchronized (deadLetters) {
+                String sequenceId = optionalSequence.get().getKey();
+                deadLetters.get(sequenceId)
+                           .addFirst(letterUpdater.apply(letter.markTouched()));
+                if (logger.isTraceEnabled()) {
+                    logger.trace("Requeued letter [{}] for sequence [{}].", letter, sequenceId);
+                }
             }
         } else {
             throw new NoSuchDeadLetterException(
@@ -186,7 +191,9 @@ public Iterable<DeadLetter<? extends M>> deadLetterSequence(@Nonnull Object sequ
     }
 
     private boolean contains(String identifier) {
-        return deadLetters.containsKey(identifier);
+        synchronized (deadLetters) {
+            return deadLetters.containsKey(identifier);
+        }
     }
 
     @Override