[Broker] Synchronize updates to the inactiveProducers map in MessageDeduplication
@@ -138,7 +138,7 @@ public class MessageDeduplication {
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);
});
@@ -169,7 +169,7 @@ public class MessageDeduplication {
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();
}
