Merge branch 'main' into fix/bubble_session_loop_exceptions

broker/src/main/java/io/moquette/broker/unsafequeues/QueuePool.java

@@ -33,7 +33,7 @@ public class QueuePool {
 
     static final boolean queueDebug = Boolean.parseBoolean(System.getProperty("moquette.queue.debug", "false"));
 
-    private static SegmentAllocationCallback callback;
+    private final SegmentAllocationCallback callback;
 
     // visible for testing
     static class SegmentRef implements Comparable<SegmentRef> {
@@ -104,6 +104,7 @@ private QueuePool(SegmentAllocator allocator, Path dataPath, int segmentSize) {
         this.allocator = allocator;
         this.dataPath = dataPath;
         this.segmentSize = segmentSize;
+        this.callback = new SegmentAllocationCallback(this);
     }
 
     private static class SegmentAllocationCallback implements PagedFilesAllocator.AllocationListener {
@@ -142,7 +143,6 @@ public static QueuePool loadQueues(Path dataPath, int pageSize, int segmentSize)
         final PagedFilesAllocator allocator = new PagedFilesAllocator(dataPath, pageSize, segmentSize, lastPage, lastSegment);
 
         final QueuePool queuePool = new QueuePool(allocator, dataPath, segmentSize);
-        callback = new SegmentAllocationCallback(queuePool);
         queuePool.loadQueueDefinitions(checkpointProps);
         LOG.debug("Loaded queues definitions: {}", queuePool.queueSegments);