[Broker] Synchronize updates to the inactiveProducers map in MessageD…

…eduplication (apache#12820)

(cherry picked from commit 9994614)
(cherry picked from commit e12bcc2)

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/MessageDeduplication.java

@@ -141,7 +141,7 @@ public MessageDeduplication(PulsarService pulsar, PersistentTopic topic, Managed
     private CompletableFuture<Void> recoverSequenceIdsMap() {
         // Load the sequence ids from the snapshot in the cursor properties
         managedCursor.getProperties().forEach((k, v) -> {
-            inactiveProducers.put(k, System.currentTimeMillis());
+            producerRemoved(k);
             highestSequencedPushed.put(k, v);
             highestSequencedPersisted.put(k, v);
         });
@@ -172,7 +172,7 @@ public void readEntriesComplete(List<Entry> entries, Object ctx) {
                     long sequenceId = Math.max(md.getHighestSequenceId(), md.getSequenceId());
                     highestSequencedPushed.put(producerName, sequenceId);
                     highestSequencedPersisted.put(producerName, sequenceId);
-                    inactiveProducers.put(producerName, System.currentTimeMillis());
+                    producerRemoved(producerName);
 
                     entry.release();
                 }