You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by cl...@apache.org on 2025/02/25 19:39:56 UTC
(activemq-artemis) 02/02: ARTEMIS-5305 Improving performance of paging with multiple producers
This is an automated email from the ASF dual-hosted git repository.
clebertsuconic pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/activemq-artemis.git
commit 24d1bbe603cadb6666a7992e296e6f94ae68e3a1
Author: Clebert Suconic <cl...@apache.org>
AuthorDate: Wed Jan 29 11:58:51 2025 -0500
ARTEMIS-5305 Improving performance of paging with multiple producers
This is improving how locking works by using a single threaded executor on paging and depaging
as well as batching writes instead of directly writing on the files from the producers.
Co-authored with a great review from Robbie Gemmel
---
.../artemis/cli/commands/tools/DBOption.java | 4 +-
.../artemis/cli/commands/tools/PrintData.java | 2 +-
.../activemq/artemis/core/paging/PagingStore.java | 30 +-
.../cursor/impl/PageCounterRebuildManager.java | 4 +-
.../paging/cursor/impl/PageCursorProviderImpl.java | 12 +-
.../activemq/artemis/core/paging/impl/Page.java | 4 +-
.../artemis/core/paging/impl/PageSyncTimer.java | 98 ----
.../artemis/core/paging/impl/PageTimedWriter.java | 200 +++++++
.../paging/impl/PagingStoreFactoryDatabase.java | 8 +-
.../core/paging/impl/PagingStoreFactoryNIO.java | 8 +-
.../artemis/core/paging/impl/PagingStoreImpl.java | 580 ++++++++++++---------
.../artemis/core/persistence/StorageManager.java | 12 +-
.../journal/AbstractJournalStorageManager.java | 11 +-
.../impl/journal/JournalStorageManager.java | 41 +-
.../impl/nullpm/NullStorageManager.java | 14 +-
.../core/replication/ReplicationManager.java | 4 +-
.../core/server/impl/ActiveMQServerImpl.java | 4 +-
.../core/server/impl/RoutingContextImpl.java | 2 +-
.../core/config/impl/ConfigurationImplTest.java | 21 +-
.../core/paging/cursor/impl/ConcurrentAckTest.java | 2 +-
.../core/server/files/FileMoveManagerTest.java | 7 +-
.../core/transaction/impl/TransactionImplTest.java | 4 +-
.../impl/PageCursorProviderTestAccessor.java | 17 +-
.../artemis/tests/util/ActiveMQTestBase.java | 55 +-
.../artemis/tests/db/paging/PageSizeTest.java | 3 +-
.../artemis/tests/db/paging/PagingTest.java | 133 ++---
.../db/paging/RealServerDatabasePagingTest.java | 3 +-
.../amqp/paging/AmqpFilterChangePageTest.java | 2 +-
.../tests/integration/client/SendAckFailTest.java | 6 +-
.../cluster/failover/BackupSyncJournalTest.java | 6 +-
.../failover/NettyReplicatedFailoverTest.java | 2 +-
.../failover/PrimaryCrashOnBackupSyncTest.java | 4 +-
.../failover/ReplicatedPagedFailoverTest.java | 2 +-
.../jms/cluster/TopicClusterPageStoreSizeTest.java | 37 +-
.../paging/GlobalJournalPagingTest.java | 2 +-
.../integration/paging/JournalPagingTest.java | 8 +-
.../paging/MessagesExpiredPagingTest.java | 4 +-
.../integration/paging/PagingReceiveTest.java | 2 +-
.../paging/TestDeadlockOnPurgePagingTest.java | 2 +-
.../integration/replication/ReplicationTest.java | 13 +-
.../SharedNothingReplicationFlowControlTest.java | 4 +-
.../artemis/tests/leak/FilteredPagingLeakTest.java | 2 +-
.../artemis/tests/leak/PagingLeakTest.java | 4 +-
.../storage/PersistMultiThreadTest.java | 31 +-
.../servers/multipleProducersPagingSoak/broker.xml | 191 +++++++
.../paging/MultipleProducersPagingSoakTest.java | 190 +++++++
.../journal/JournalStorageManagerAccessor.java} | 16 +-
.../tests/unit/core/paging/impl/AmqpPageTest.java | 4 +-
.../tests/unit/core/paging/impl/PageTest.java | 4 +-
.../core/paging/impl/PageTimedWriterUnitTest.java | 532 +++++++++++++++++++
.../core/paging/impl/PagingManagerImplTest.java | 27 +-
.../unit/core/paging/impl/PagingStoreImplTest.java | 392 ++++----------
52 files changed, 1870 insertions(+), 900 deletions(-)
diff --git a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/DBOption.java b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/DBOption.java
index 837df21a33..a437e6c24d 100644
--- a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/DBOption.java
+++ b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/DBOption.java
@@ -301,12 +301,12 @@ public class DBOption extends OptionalLocking {
PagingStoreFactory pageStoreFactory = new PagingStoreFactoryDatabase((DatabaseStorageConfiguration) configuration.getStoreConfiguration(),
storageManager, 1000L,
- scheduledExecutorService, executorFactory, executorFactory,
+ scheduledExecutorService, executorFactory,
false, null);
pagingmanager = new PagingManagerImpl(pageStoreFactory, addressSettingsRepository, configuration.getManagementAddress());
} else {
storageManager = new JournalStorageManager(config, EmptyCriticalAnalyzer.getInstance(), executorFactory, executorFactory);
- PagingStoreFactory pageStoreFactory = new PagingStoreFactoryNIO(storageManager, config.getPagingLocation(), 1000L, scheduledExecutorService, executorFactory, executorFactory, true, null);
+ PagingStoreFactory pageStoreFactory = new PagingStoreFactoryNIO(storageManager, config.getPagingLocation(), 1000L, scheduledExecutorService, executorFactory, true, null);
pagingmanager = new PagingManagerImpl(pageStoreFactory, addressSettingsRepository, configuration.getManagementAddress());
}
diff --git a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/PrintData.java b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/PrintData.java
index 7ad15716cb..055d9bce86 100644
--- a/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/PrintData.java
+++ b/artemis-cli/src/main/java/org/apache/activemq/artemis/cli/commands/tools/PrintData.java
@@ -233,7 +233,7 @@ public class PrintData extends DBOption {
try {
final StorageManager sm = new NullStorageManager();
- PagingStoreFactory pageStoreFactory = new PagingStoreFactoryNIO(sm, pageDirectory, 1000L, scheduled, execfactory, execfactory, false, null);
+ PagingStoreFactory pageStoreFactory = new PagingStoreFactoryNIO(sm, pageDirectory, 1000L, scheduled, execfactory, false, null);
HierarchicalRepository<AddressSettings> addressSettingsRepository = new HierarchicalObjectRepository<>();
addressSettingsRepository.setDefault(new AddressSettings());
PagingManager manager = new PagingManagerImpl(pageStoreFactory, addressSettingsRepository);
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagingStore.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagingStore.java
index 184bc539ea..1dd8d5ac14 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagingStore.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/PagingStore.java
@@ -27,7 +27,6 @@ import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.core.paging.cursor.PageCursorProvider;
import org.apache.activemq.artemis.core.paging.cursor.PageSubscription;
import org.apache.activemq.artemis.core.paging.impl.Page;
-import org.apache.activemq.artemis.core.persistence.OperationContext;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.replication.ReplicationManager;
import org.apache.activemq.artemis.core.server.ActiveMQComponent;
@@ -114,11 +113,6 @@ public interface PagingStore extends ActiveMQComponent, RefCountMessageListener
* It's ok to look for this with an estimate on starting a task or not, but you will need to recheck on actual paging operations. */
boolean isPaging();
- /**
- * Schedules sync to the file storage.
- */
- void addSyncPoint(OperationContext context) throws Exception;
-
/**
* Performs a real sync on the current IO file.
*/
@@ -162,7 +156,11 @@ public interface PagingStore extends ActiveMQComponent, RefCountMessageListener
Page removePage(int pageId);
- void forceAnotherPage() throws Exception;
+ default void forceAnotherPage() throws Exception {
+ forceAnotherPage(false);
+ }
+
+ void forceAnotherPage(boolean useExecutor) throws Exception;
Page getCurrentPage();
@@ -204,6 +202,8 @@ public interface PagingStore extends ActiveMQComponent, RefCountMessageListener
*/
boolean checkReleasedMemory();
+ void writeLock();
+
/**
* Write lock the PagingStore.
*
@@ -211,12 +211,18 @@ public interface PagingStore extends ActiveMQComponent, RefCountMessageListener
* indefinitely.
* @return {@code true} if the lock was obtained, {@code false} otherwise
*/
- boolean lock(long timeout);
+ boolean writeLock(long timeout);
/**
- * Releases locks acquired with {@link PagingStore#lock(long)}.
+ * Releases locks acquired with {@link PagingStore#writeLock(long)}.
*/
- void unlock();
+ void writeUnlock();
+
+ void readLock();
+
+ boolean readLock(long timeout);
+
+ void readUnlock();
/**
* This is used mostly by tests.
@@ -265,6 +271,10 @@ public interface PagingStore extends ActiveMQComponent, RefCountMessageListener
void unblock();
+ default boolean hasPendingIO() {
+ return false;
+ }
+
default StorageManager getStorageManager() {
return null;
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCounterRebuildManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCounterRebuildManager.java
index 6371c7001a..1311ce07ac 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCounterRebuildManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCounterRebuildManager.java
@@ -71,7 +71,7 @@ public class PageCounterRebuildManager implements Runnable {
/** this method will perform the copy from Acked recorded from the subscription into a separate data structure.
* So we can count data while we consolidate at the end */
private void initialize(PagingStore store) {
- store.lock(-1);
+ store.writeLock();
try {
try {
paging = store.isPaging();
@@ -134,7 +134,7 @@ public class PageCounterRebuildManager implements Runnable {
});
});
} finally {
- store.unlock();
+ store.writeUnlock();
}
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java
index 481ade2d6b..89a44e3ebb 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderImpl.java
@@ -275,7 +275,7 @@ public class PageCursorProviderImpl implements PageCursorProvider {
try (ArtemisCloseable readLock = storageManager.closeableReadLock()) {
while (true) {
- if (pagingStore.lock(100)) {
+ if (pagingStore.writeLock(1_000)) {
break;
}
if (!pagingStore.isStarted())
@@ -318,8 +318,8 @@ public class PageCursorProviderImpl implements PageCursorProvider {
assert pagingStore.getNumberOfPages() >= 0;
- if (pagingStore.getNumberOfPages() == 0 || pagingStore.getNumberOfPages() == 1 && (pagingStore.getCurrentPage() == null || pagingStore.getCurrentPage().getNumberOfMessages() == 0)) {
- logger.trace("StopPaging being called on {}", pagingStore);
+ if (!pagingStore.hasPendingIO() && (pagingStore.getNumberOfPages() == 0 || pagingStore.getNumberOfPages() == 1 && (pagingStore.getCurrentPage() == null || pagingStore.getCurrentPage().getNumberOfMessages() == 0))) {
+ logger.trace("StopPaging being called on {}, pending={}", pagingStore, pagingStore.hasPendingIO());
pagingStore.stopPaging();
} else {
if (logger.isTraceEnabled()) {
@@ -333,7 +333,7 @@ public class PageCursorProviderImpl implements PageCursorProvider {
return;
} finally {
logger.trace("<<<< Cleanup end on {}", pagingStore.getAddress());
- pagingStore.unlock();
+ pagingStore.writeUnlock();
}
}
}
@@ -452,7 +452,9 @@ public class PageCursorProviderImpl implements PageCursorProvider {
storeBookmark(cursorList, currentPage);
- pagingStore.stopPaging();
+ if (!pagingStore.hasPendingIO()) {
+ pagingStore.stopPaging();
+ }
}
// Protected as a way to inject testing
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/Page.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/Page.java
index ea98c8a1c0..2ca9c47646 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/Page.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/Page.java
@@ -178,9 +178,9 @@ public final class Page {
return sb.toString();
}
- public synchronized void write(final PagedMessage message) throws Exception {
+ public synchronized void write(final PagedMessage message, boolean lineUp, boolean originallyReplicated) throws Exception {
writeDirect(message);
- storageManager.pageWrite(storeName, message, pageId);
+ storageManager.pageWrite(storeName, message, pageId, lineUp, originallyReplicated);
}
/** This write will not interact back with the storage manager.
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PageSyncTimer.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PageSyncTimer.java
deleted file mode 100644
index 38e613a1aa..0000000000
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PageSyncTimer.java
+++ /dev/null
@@ -1,98 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.activemq.artemis.core.paging.impl;
-
-import java.util.LinkedList;
-import java.util.List;
-import java.util.concurrent.Executor;
-import java.util.concurrent.ScheduledExecutorService;
-import java.util.concurrent.TimeUnit;
-
-import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
-import org.apache.activemq.artemis.core.paging.PagingStore;
-import org.apache.activemq.artemis.core.persistence.OperationContext;
-import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
-
-/**
- * This will batch multiple calls waiting to perform a sync in a single call.
- */
-final class PageSyncTimer extends ActiveMQScheduledComponent {
-
-
-
- private final PagingStore store;
-
- private final ScheduledExecutorService scheduledExecutor;
-
- private boolean pendingSync;
-
- private final long timeSync;
-
- private final List<OperationContext> syncOperations = new LinkedList<>();
-
-
- PageSyncTimer(PagingStore store, ScheduledExecutorService scheduledExecutor, Executor executor, long timeSync) {
- super(scheduledExecutor, executor, timeSync, TimeUnit.NANOSECONDS, true);
- this.store = store;
- this.scheduledExecutor = scheduledExecutor;
- this.timeSync = timeSync;
- }
-
-
- synchronized void addSync(OperationContext ctx) {
- ctx.pageSyncLineUp();
- if (!pendingSync) {
- pendingSync = true;
-
- delay();
- }
- syncOperations.add(ctx);
- }
-
- @Override
- public void run() {
- tick();
- }
-
- private void tick() {
- OperationContext[] pendingSyncsArray;
- synchronized (this) {
-
- pendingSync = false;
- pendingSyncsArray = new OperationContext[syncOperations.size()];
- pendingSyncsArray = syncOperations.toArray(pendingSyncsArray);
- syncOperations.clear();
- }
-
- try {
- if (pendingSyncsArray.length != 0) {
- store.ioSync();
- }
- } catch (Exception e) {
- for (OperationContext ctx : pendingSyncsArray) {
- ctx.onError(ActiveMQExceptionType.IO_ERROR.getCode(), e.getClass() + " during ioSync for paging on " + store.getStoreName() + ": " + e.getMessage());
- }
- } finally {
- // In case of failure, The context should propagate an exception to the client
- // We send an exception to the client even on the case of a failure
- // to avoid possible locks and the client not getting the exception back
- for (OperationContext ctx : pendingSyncsArray) {
- ctx.pageSyncDone();
- }
- }
- }
-}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PageTimedWriter.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PageTimedWriter.java
new file mode 100644
index 0000000000..7fbd8fac47
--- /dev/null
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PageTimedWriter.java
@@ -0,0 +1,200 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.artemis.core.paging.impl;
+
+import java.lang.invoke.MethodHandles;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
+import org.apache.activemq.artemis.api.core.ActiveMQExceptionType;
+import org.apache.activemq.artemis.core.paging.PagedMessage;
+import org.apache.activemq.artemis.core.persistence.OperationContext;
+import org.apache.activemq.artemis.core.persistence.StorageManager;
+import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
+import org.apache.activemq.artemis.core.server.ActiveMQScheduledComponent;
+import org.apache.activemq.artemis.core.server.RouteContextList;
+import org.apache.activemq.artemis.core.transaction.Transaction;
+import org.apache.activemq.artemis.utils.ArtemisCloseable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class PageTimedWriter extends ActiveMQScheduledComponent {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private final PagingStoreImpl pagingStore;
+
+ private final StorageManager storageManager;
+
+ protected final List<PageEvent> pageEvents = new ArrayList<>();
+
+ protected volatile int pendingTasks = 0;
+
+ protected final boolean syncNonTX;
+
+ private final Semaphore writeCredits;
+
+ private final int maxCredits;
+
+ private static final AtomicIntegerFieldUpdater<PageTimedWriter> pendingTasksUpdater = AtomicIntegerFieldUpdater.newUpdater(PageTimedWriter.class, "pendingTasks");
+
+ public boolean hasPendingIO() {
+ return pendingTasksUpdater.get(this) > 0;
+ }
+
+ public static class PageEvent {
+
+ PageEvent(OperationContext context, PagedMessage message, Transaction tx, RouteContextList listCtx, int credits, boolean replicated) {
+ this.context = context;
+ this.message = message;
+ this.listCtx = listCtx;
+ this.replicated = replicated;
+ this.credits = credits;
+ this.tx = tx;
+ }
+
+ final boolean replicated;
+ final PagedMessage message;
+ final OperationContext context;
+ final RouteContextList listCtx;
+ final Transaction tx;
+ final int credits;
+ }
+
+ public PageTimedWriter(int writeCredits, StorageManager storageManager, PagingStoreImpl pagingStore, ScheduledExecutorService scheduledExecutor, Executor executor, boolean syncNonTX, long timeSync) {
+ super(scheduledExecutor, executor, timeSync, TimeUnit.NANOSECONDS, true);
+ this.pagingStore = pagingStore;
+ this.storageManager = storageManager;
+ this.syncNonTX = syncNonTX;
+ this.writeCredits = new Semaphore(writeCredits);
+ this.maxCredits = writeCredits;
+ }
+
+ public int getMaxCredits() {
+ return maxCredits;
+ }
+
+ @Override
+ public synchronized void stop() {
+ super.stop();
+ processMessages();
+ }
+
+ /** We increment task while holding the readLock.
+ * This is because we verify if the system is paging, and we get out of paging when no pending tasks and no pending messages.
+ * We allocate a task while holding the read Lock.
+ * We cannot call addTask within the lock as if the semaphore gets out of credits we would deadlock in certain cases. */
+ public void incrementTask() {
+ pendingTasksUpdater.incrementAndGet(this);
+ }
+
+ public void addTask(OperationContext context,
+ PagedMessage message,
+ Transaction tx,
+ RouteContextList listCtx) {
+
+ if (!isStarted()) {
+ throw new IllegalStateException("PageWriter Service is stopped");
+ }
+ int credits = Math.min(message.getEncodeSize() + PageReadWriter.SIZE_RECORD, maxCredits);
+ writeCredits.acquireUninterruptibly(credits);
+ synchronized (this) {
+ final boolean replicated = storageManager.isReplicated();
+ PageEvent event = new PageEvent(context, message, tx, listCtx, credits, replicated);
+ context.storeLineUp();
+ if (replicated) {
+ context.replicationLineUp();
+ }
+ this.pageEvents.add(event);
+ delay();
+ }
+
+ }
+
+ private synchronized PageEvent[] extractPendingEvents() {
+ if (pageEvents.isEmpty()) {
+ return null;
+ }
+ PageEvent[] pendingsWrites = new PageEvent[pageEvents.size()];
+ pendingsWrites = pageEvents.toArray(pendingsWrites);
+ pageEvents.clear();
+ return pendingsWrites;
+ }
+
+ @Override
+ public void run() {
+ ArtemisCloseable closeable = storageManager.closeableReadLock(true);
+ if (closeable == null) {
+ logger.trace("Delaying PagedTimedWriter as it's currently locked");
+ delay();
+ } else {
+ try {
+ processMessages();
+ } finally {
+ closeable.close();
+ }
+ }
+ }
+
+ protected void processMessages() {
+ PageEvent[] pendingEvents = extractPendingEvents();
+ if (pendingEvents == null) {
+ return;
+ }
+ OperationContext beforeContext = OperationContextImpl.getContext();
+
+ try {
+ boolean requireSync = false;
+ for (PageEvent event : pendingEvents) {
+ OperationContextImpl.setContext(event.context);
+ pagingStore.directWritePage(event.message, false, event.replicated);
+
+ if (event.tx != null || syncNonTX) {
+ requireSync = true;
+ }
+ }
+ if (requireSync) {
+ performSync();
+ }
+
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ // In case of failure, The context should propagate an exception to the client
+ // We send an exception to the client even on the case of a failure
+ // to avoid possible locks and the client not getting the exception back
+ for (PageEvent event : pendingEvents) {
+ event.context.onError(ActiveMQExceptionType.IO_ERROR.getCode(), e.getClass() + " during ioSync for paging on " + pagingStore.getStoreName() + ": " + e.getMessage());
+ }
+ } finally {
+ for (PageEvent event : pendingEvents) {
+ event.context.done();
+ pendingTasksUpdater.decrementAndGet(this);
+ writeCredits.release(event.credits);
+ }
+ OperationContextImpl.setContext(beforeContext);
+ }
+ }
+
+ protected void performSync() throws Exception {
+ pagingStore.ioSync();
+ }
+}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryDatabase.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryDatabase.java
index d5c19bcaff..69710d5cc4 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryDatabase.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryDatabase.java
@@ -73,8 +73,6 @@ public class PagingStoreFactoryDatabase implements PagingStoreFactory {
private ExecutorFactory executorFactory;
- private ExecutorFactory ioExecutorFactory;
-
private JDBCSequentialFileFactory pagingFactoryFileFactory;
@Override
@@ -98,12 +96,10 @@ public class PagingStoreFactoryDatabase implements PagingStoreFactory {
final long syncTimeout,
final ScheduledExecutorService scheduledExecutor,
final ExecutorFactory executorFactory,
- final ExecutorFactory ioExecutorFactory,
final boolean syncNonTransactional,
final IOCriticalErrorListener criticalErrorListener) throws Exception {
this.storageManager = storageManager;
this.executorFactory = executorFactory;
- this.ioExecutorFactory = ioExecutorFactory;
this.syncNonTransactional = syncNonTransactional;
this.scheduledExecutor = scheduledExecutor;
this.syncTimeout = syncTimeout;
@@ -153,7 +149,7 @@ public class PagingStoreFactoryDatabase implements PagingStoreFactory {
@Override
public synchronized PagingStore newStore(final SimpleString address, final AddressSettings settings) {
- return new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, null, this, address, settings, executorFactory.getExecutor().setFair(true), ioExecutorFactory.getExecutor(), syncNonTransactional);
+ return new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, null, this, address, settings, executorFactory.getExecutor().setFair(true), syncNonTransactional);
}
@Override
@@ -233,7 +229,7 @@ public class PagingStoreFactoryDatabase implements PagingStoreFactory {
AddressSettings settings = addressSettingsRepository.getMatch(address.toString());
- PagingStore store = new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, factory, this, address, settings, executorFactory.getExecutor().setFair(true), ioExecutorFactory.getExecutor(), syncNonTransactional);
+ PagingStore store = new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, factory, this, address, settings, executorFactory.getExecutor().setFair(true), syncNonTransactional);
storesReturn.add(store);
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryNIO.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryNIO.java
index 5783488d36..20dbf7b5c7 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryNIO.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreFactoryNIO.java
@@ -62,8 +62,6 @@ public class PagingStoreFactoryNIO implements PagingStoreFactory {
private final ExecutorFactory executorFactory;
- private final ExecutorFactory ioExecutorFactory;
-
private final boolean syncNonTransactional;
private PagingManager pagingManager;
@@ -109,13 +107,11 @@ public class PagingStoreFactoryNIO implements PagingStoreFactory {
final long syncTimeout,
final ScheduledExecutorService scheduledExecutor,
final ExecutorFactory executorFactory,
- final ExecutorFactory ioExecutorFactory,
final boolean syncNonTransactional,
final IOCriticalErrorListener critialErrorListener) {
this.storageManager = storageManager;
this.directory = directory;
this.executorFactory = executorFactory;
- this.ioExecutorFactory = ioExecutorFactory;
this.syncNonTransactional = syncNonTransactional;
this.scheduledExecutor = scheduledExecutor;
this.syncTimeout = syncTimeout;
@@ -153,7 +149,7 @@ public class PagingStoreFactoryNIO implements PagingStoreFactory {
@Override
public synchronized PagingStore newStore(final SimpleString address, final AddressSettings settings) {
- return new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, null, this, address, settings, executorFactory.getExecutor().setFair(true), ioExecutorFactory.getExecutor(), syncNonTransactional);
+ return new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, null, this, address, settings, executorFactory.getExecutor().setFair(true), syncNonTransactional);
}
@Override
@@ -230,7 +226,7 @@ public class PagingStoreFactoryNIO implements PagingStoreFactory {
AddressSettings settings = addressSettingsRepository.getMatch(address.toString());
- PagingStore store = new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, factory, this, address, settings, executorFactory.getExecutor(), executorFactory.getExecutor(), syncNonTransactional);
+ PagingStore store = new PagingStoreImpl(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, factory, this, address, settings, executorFactory.getExecutor(), syncNonTransactional);
storesReturn.add(store);
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
index 432a0eb190..347258c131 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
@@ -62,6 +62,7 @@ import org.apache.activemq.artemis.core.transaction.TransactionOperation;
import org.apache.activemq.artemis.core.transaction.TransactionPropertyIndexes;
import org.apache.activemq.artemis.utils.ArtemisCloseable;
import org.apache.activemq.artemis.utils.FutureLatch;
+import org.apache.activemq.artemis.utils.SimpleFutureImpl;
import org.apache.activemq.artemis.utils.SizeAwareMetric;
import org.apache.activemq.artemis.utils.actors.ArtemisExecutor;
import org.apache.activemq.artemis.utils.runnables.AtomicRunnable;
@@ -85,7 +86,7 @@ public class PagingStoreImpl implements PagingStore {
private final PageCache usedPages = new PageCache(this);
- //it's being guarded by lock.writeLock().lock() and never read concurrently
+ // This is updated and read by the Page's executor thread
private long currentPageSize = 0;
private final SimpleString storeName;
@@ -95,8 +96,8 @@ public class PagingStoreImpl implements PagingStore {
private final PagingStoreFactory storeFactory;
- // Used to schedule sync threads
- private final PageSyncTimer syncTimer;
+ // this is used to batch and sync into paging asynchronously
+ private final PageTimedWriter timedWriter;
private long maxSize;
@@ -155,6 +156,8 @@ public class PagingStoreImpl implements PagingStore {
private final PageCursorProvider cursorProvider;
+ // This lock mostly protects the paging field.
+ // It is also used to block producers in eventual cases such as dropping a queue, but mostly to protect if the storage is in paging mode
private final ReadWriteLock lock = new ReentrantReadWriteLock();
private volatile boolean running = false;
@@ -177,8 +180,10 @@ public class PagingStoreImpl implements PagingStore {
final SimpleString storeName,
final AddressSettings addressSettings,
final ArtemisExecutor executor,
- final ArtemisExecutor ioExecutor,
final boolean syncNonTransactional) {
+ if (scheduledExecutor == null) {
+ throw new NullPointerException("scheduledExecutor = null");
+ }
if (pagingManager == null) {
throw new IllegalStateException("Paging Manager can't be null");
}
@@ -205,17 +210,28 @@ public class PagingStoreImpl implements PagingStore {
this.syncNonTransactional = syncNonTransactional;
- if (scheduledExecutor != null && syncTimeout > 0) {
- this.syncTimer = new PageSyncTimer(this, scheduledExecutor, ioExecutor, syncTimeout);
- } else {
- this.syncTimer = null;
- }
+ this.timedWriter = createPageTimedWriter(scheduledExecutor, syncTimeout);
this.cursorProvider = storeFactory.newCursorProvider(this, this.storageManager, addressSettings, executor);
this.usingGlobalMaxSize = pagingManager.isUsingGlobalSize();
}
+ // This is an extension point for unit tests to replace the creation of the PagedTimeWriter
+ protected PageTimedWriter createPageTimedWriter(ScheduledExecutorService scheduledExecutor, long syncTimeout) {
+ if (scheduledExecutor == null) {
+ throw new NullPointerException("scheduledExecutor");
+ }
+ if (executor == null) {
+ throw new NullPointerException("executor");
+ }
+ // Notice that any calls on the PageTimedWriter are going to use the paging store's executor.
+ // The scheduledExecutor will transfer the call to the paging store executor.
+ PageTimedWriter localWriter = new PageTimedWriter(pageSize, storageManager, this, scheduledExecutor, executor, syncNonTransactional, syncTimeout);
+ localWriter.start();
+ return localWriter;
+ }
+
private void overSized() {
full = true;
}
@@ -391,21 +407,85 @@ public class PagingStoreImpl implements PagingStore {
}
}
+
+ @Override
+ public void readLock() {
+ readLock(-1L);
+ }
+
@Override
- public boolean lock(long timeout) {
- if (timeout == -1) {
- lock.writeLock().lock();
+ public boolean readLock(long timeout) {
+ try {
+ if (timeout == -1) {
+ while (true) {
+ if (tryReadLock(1, TimeUnit.SECONDS)) {
+ return true;
+ }
+ }
+ } else {
+ return tryReadLock(timeout, TimeUnit.MILLISECONDS);
+ }
+ } catch (InterruptedException e) {
+ logger.warn(e.getMessage(), e);
+ Thread.currentThread().interrupt();
+ return false;
+ }
+ }
+
+ private boolean tryReadLock(long timeout, TimeUnit unit) throws InterruptedException {
+ if (lock.readLock().tryLock(timeout, unit)) {
return true;
+ } else {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Not able to read lock");
+ }
+ return false;
}
+ }
+
+ @Override
+ public void readUnlock() {
+ lock.readLock().unlock();
+ }
+
+ @Override
+ public void writeLock() {
+ writeLock(-1L);
+ }
+
+ @Override
+ public boolean writeLock(long timeout) {
try {
- return lock.writeLock().tryLock(timeout, TimeUnit.MILLISECONDS);
+ if (timeout == -1) {
+ while (true) {
+ if (tryWriteLock(1, TimeUnit.SECONDS)) {
+ return true;
+ }
+ }
+ } else {
+ return tryWriteLock(timeout, TimeUnit.MILLISECONDS);
+ }
} catch (InterruptedException e) {
+ logger.warn(e.getMessage(), e);
+ Thread.currentThread().interrupt();
return false;
}
}
+ private boolean tryWriteLock(long timeout, TimeUnit unit) throws InterruptedException {
+ if (lock.writeLock().tryLock(timeout, unit)) {
+ return true;
+ } else {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Not able to write lock");
+ }
+ return false;
+ }
+
+ }
+
@Override
- public void unlock() {
+ public void writeUnlock() {
lock.writeLock().unlock();
}
@@ -526,29 +606,10 @@ public class PagingStoreImpl implements PagingStore {
return storeName;
}
- @Override
- public void addSyncPoint(OperationContext context) throws Exception {
- if (fileFactory == null || !fileFactory.supportsIndividualContext()) {
- if (syncTimer != null) {
- syncTimer.addSync(context);
- } else {
- ioSync();
- }
- }
- }
-
@Override
public void ioSync() throws Exception {
if (!fileFactory.supportsIndividualContext()) {
- Page page;
- lock.readLock().lock();
-
- try {
- page = currentPage;
- } finally {
- lock.readLock().unlock();
- }
-
+ Page page = currentPage;
if (page != null) {
page.trySync();
}
@@ -631,7 +692,7 @@ public class PagingStoreImpl implements PagingStore {
@Override
public void start() throws Exception {
- lock.writeLock().lock();
+ writeLock();
try {
@@ -688,7 +749,7 @@ public class PagingStoreImpl implements PagingStore {
}
} finally {
- lock.writeLock().unlock();
+ writeUnlock();
}
}
@@ -731,17 +792,23 @@ public class PagingStoreImpl implements PagingStore {
@Override
public void stopPaging() {
logger.debug("stopPaging being called, while isPaging={} on {}", this.paging, this.storeName);
- lock.writeLock().lock();
+ writeLock();
try {
final boolean isPaging = this.paging;
if (isPaging) {
+ assert !timedWriter.hasPendingIO() : "There is pending IO on PagingStoreImpl while calling stopPaging";
+ if (timedWriter.hasPendingIO()) {
+ // it should not happen, however if it happened we will just ignore the call for stopPaging
+ logger.debug("There are pending timed writes. Cannot clear paging now.");
+ return;
+ }
paging = false;
ActiveMQServerLogger.LOGGER.pageStoreStop(storeName, getPageInfo());
pageLimitReleased();
}
this.cursorProvider.onPageModeCleared();
} finally {
- lock.writeLock().unlock();
+ writeUnlock();
}
}
@@ -755,47 +822,52 @@ public class PagingStoreImpl implements PagingStore {
return false;
}
- lock.readLock().lock();
+ readLock();
try {
// I'm not calling isPaging() here because i need to be atomic and hold a lock.
if (paging) {
return false;
}
} finally {
- lock.readLock().unlock();
+ readUnlock();
}
- // if the first check failed, we do it again under a global currentPageLock
- // (writeLock) this time
- lock.writeLock().lock();
-
- try {
- // Same notes from previous if (paging) on this method will apply here
- if (paging) {
- return false;
- }
-
+ // We need to guarantee a readLock on the storageManager before starting paging.
+ // This is because the replication manager will get a list of files to synchronize while
+ // holding a writeLock on the storageManager.
+ // So we must guarantee a readLock here otherwise the list might be wrong.
+ try (ArtemisCloseable readLock = storageManager.closeableReadLock()) {
+ // if the first check failed, we do it again under a global currentPageLock
+ // (writeLock) this time
+ writeLock();
try {
- if (currentPage == null) {
- openNewPage();
- } else {
- if (!currentPage.getFile().exists() || !currentPage.getFile().isOpen()) {
- currentPage.getFile().open();
+ // Same notes from previous if (paging) on this method will apply here
+ if (paging) {
+ return false;
+ }
+
+ try {
+ if (currentPage == null) {
+ openNewPage();
+ } else {
+ if (!currentPage.getFile().exists() || !currentPage.getFile().isOpen()) {
+ currentPage.getFile().open();
+ }
}
+ } catch (Exception e) {
+ // If not possible to starting page due to an IO error, we will just consider it non paging.
+ // This shouldn't happen anyway
+ ActiveMQServerLogger.LOGGER.pageStoreStartIOError(e);
+ storageManager.criticalError(e);
+ return false;
}
- } catch (Exception e) {
- // If not possible to starting page due to an IO error, we will just consider it non paging.
- // This shouldn't happen anyway
- ActiveMQServerLogger.LOGGER.pageStoreStartIOError(e);
- storageManager.criticalError(e);
- return false;
- }
- paging = true;
- ActiveMQServerLogger.LOGGER.pageStoreStart(storeName, getPageInfo());
+ paging = true;
+ ActiveMQServerLogger.LOGGER.pageStoreStart(storeName, getPageInfo());
- return true;
- } finally {
- lock.writeLock().unlock();
+ return true;
+ } finally {
+ writeUnlock();
+ }
}
}
@@ -894,7 +966,28 @@ public class PagingStoreImpl implements PagingStore {
@Override
public void forceAnotherPage() throws Exception {
- openNewPage();
+ forceAnotherPage(false);
+ }
+
+ @Override
+ public void forceAnotherPage(boolean useExecutor) throws Exception {
+ // we need to open a new page inside the executor
+ // as the PageTimedWriter will write on the currentPage without holding a writeLock
+ // the current page must be changed within the executor's
+ if (useExecutor) {
+ SimpleFutureImpl future = new SimpleFutureImpl();
+ execute(() -> {
+ try {
+ openNewPage();
+ future.set(true);
+ } catch (Exception e) {
+ future.fail(e);
+ }
+ });
+ future.get();
+ } else {
+ openNewPage();
+ }
}
@@ -913,50 +1006,45 @@ public class PagingStoreImpl implements PagingStore {
@Override
public Page removePage(int pageId) {
try {
- lock.writeLock().lock(); // Make sure no checks are done on currentPage while we are depaging
- try {
- if (!running) {
- return null;
- }
+ if (!running) {
+ return null;
+ }
- if (currentPageId == pageId) {
- if (logger.isDebugEnabled()) {
- logger.debug("Ignoring remove({}) as this is the current writing page", pageId);
- }
- // we don't deal with the current page, we let that one to be cleared from the regular depage
- return null;
+ if (currentPageId == pageId) {
+ if (logger.isDebugEnabled()) {
+ logger.debug("Ignoring remove({}) as this is the current writing page", pageId);
}
+ // we don't deal with the current page, we let that one to be cleared from the regular depage
+ return null;
+ }
- Page page = usePage(pageId, false);
+ Page page = usePage(pageId, false);
- if (page == null) {
- page = newPageObject(pageId);
- }
+ if (page == null) {
+ page = newPageObject(pageId);
+ }
- if (page != null && page.getFile().exists()) {
- page.usageDown();
- // we only decrement numberOfPages if the file existed
- // it could have been removed by a previous delete
- // on this case we just need to ignore this and move on
- numberOfPages--;
- }
+ if (page != null && page.getFile().exists()) {
+ page.usageDown();
+ // we only decrement numberOfPages if the file existed
+ // it could have been removed by a previous delete
+ // on this case we just need to ignore this and move on
+ numberOfPages--;
+ }
- if (logger.isTraceEnabled()) {
- logger.trace("Removing page {}, now containing numberOfPages={}", pageId, numberOfPages);
- }
+ if (logger.isTraceEnabled()) {
+ logger.trace("Removing page {}, now containing numberOfPages={}", pageId, numberOfPages);
+ }
- if (numberOfPages == 0) {
- if (logger.isTraceEnabled()) {
- logger.trace("Page has no pages after removing last page {}", pageId, new Exception("Trace"));
- }
+ if (numberOfPages == 0) {
+ if (logger.isTraceEnabled()) {
+ logger.trace("Page has no pages after removing last page {}", pageId, new Exception("Trace"));
}
+ }
- assert numberOfPages >= 0 : "numberOfPages should never be negative. on removePage(" + pageId + "). numberOfPages=" + numberOfPages;
+ assert numberOfPages >= 0 : "numberOfPages should never be negative. on removePage(" + pageId + "). numberOfPages=" + numberOfPages;
- return page;
- } finally {
- lock.writeLock().unlock();
- }
+ return page;
} catch (Throwable e) {
logger.warn(e.getMessage(), e);
if (e instanceof AssertionError error) {
@@ -982,80 +1070,74 @@ public class PagingStoreImpl implements PagingStore {
*/
@Override
public Page depage() throws Exception {
- lock.writeLock().lock(); // Make sure no checks are done on currentPage while we are depaging
- try {
- if (!running) {
- return null;
- }
+ if (!running) {
+ return null;
+ }
- if (numberOfPages == 0) {
- return null;
- } else {
- final Page returnPage;
+ if (numberOfPages == 0) {
+ return null;
+ } else {
+ final Page returnPage;
- numberOfPages--;
+ numberOfPages--;
- // We are out of old pages, all that is left now is the current page.
- // On that case we need to replace it by a new empty page, and return the current page immediately
- if (currentPageId == firstPageId) {
- firstPageId = Integer.MAX_VALUE;
- logger.trace("Setting up firstPageID=MAX_VALUE");
+ // We are out of old pages, all that is left now is the current page.
+ // On that case we need to replace it by a new empty page, and return the current page immediately
+ if (currentPageId == firstPageId) {
+ firstPageId = Integer.MAX_VALUE;
+ logger.trace("Setting up firstPageID=MAX_VALUE");
- if (currentPage == null) {
- // sanity check... it shouldn't happen!
- throw new IllegalStateException("CurrentPage is null");
- }
+ if (currentPage == null) {
+ // sanity check... it shouldn't happen!
+ throw new IllegalStateException("CurrentPage is null");
+ }
- returnPage = currentPage;
- returnPage.close(true);
- resetCurrentPage(null);
+ returnPage = currentPage;
+ returnPage.close(true);
+ resetCurrentPage(null);
- // The current page is empty... which means we reached the end of the pages
- if (returnPage.getNumberOfMessages() == 0) {
- stopPaging();
- returnPage.open(true);
- returnPage.delete(null);
+ // The current page is empty... which means we reached the end of the pages
+ if (returnPage.getNumberOfMessages() == 0 && !hasPendingIO()) {
+ stopPaging();
+ returnPage.open(true);
+ returnPage.delete(null);
- // This will trigger this address to exit the page mode,
- // and this will make ActiveMQ Artemis start using the journal again
- return null;
- } else {
- // We need to create a new page, as we can't lock the address until we finish depaging.
- openNewPage();
- }
+ // This will trigger this address to exit the page mode,
+ // and this will make ActiveMQ Artemis start using the journal again
+ return null;
} else {
- if (logger.isTraceEnabled()) {
- logger.trace("firstPageId++ = beforeIncrement={}", firstPageId);
- }
- long pageNR = firstPageId++;
-
- // first we look for the page on the used Pages cache
- // if non existing, we just create a new one outside of the cache
- // as we should not introduce any extras
- Page usedPage = usePage(pageNR, false);
- if (usedPage == null) {
- returnPage = newPageObject(pageNR);
- } else {
- returnPage = usedPage;
- }
+ // We need to create a new page, as we can't lock the address until we finish depaging.
+ openNewPage();
}
-
- if (!returnPage.getFile().exists()) {
- // if the file does not exist, we will just increment back to where it was before
- numberOfPages++;
+ } else {
+ if (logger.isTraceEnabled()) {
+ logger.trace("firstPageId++ = beforeIncrement={}", firstPageId);
}
+ long pageNR = firstPageId++;
+
+ // first we look for the page on the used Pages cache
+ // if non existing, we just create a new one outside of the cache
+ // as we should not introduce any extras
+ Page usedPage = usePage(pageNR, false);
+ if (usedPage == null) {
+ returnPage = newPageObject(pageNR);
+ } else {
+ returnPage = usedPage;
+ }
+ }
- // we make this assertion after checking the file existed before.
- // this could be eventually negative for a short period of time
- // but after compensating the non existent file the assertion should still hold true
- assert numberOfPages >= 0 : "numberOfPages should never be negative. on depage(). currentPageId=" + currentPageId + ", firstPageId=" + firstPageId + "";
-
- return returnPage;
+ if (!returnPage.getFile().exists()) {
+ // if the file does not exist, we will just increment back to where it was before
+ numberOfPages++;
}
- } finally {
- lock.writeLock().unlock();
- }
+ // we make this assertion after checking the file existed before.
+ // this could be eventually negative for a short period of time
+ // but after compensating the non existent file the assertion should still hold true
+ assert numberOfPages >= 0 : "numberOfPages should never be negative. on depage(). currentPageId=" + currentPageId + ", firstPageId=" + firstPageId + "";
+
+ return returnPage;
+ }
}
private final Queue<Runnable> onMemoryFreedRunnables = new ConcurrentLinkedQueue<>();
@@ -1224,18 +1306,6 @@ public class PagingStoreImpl implements PagingStore {
return false;
}
- // We need to ensure a read lock, as depage could change the paging state
- lock.readLock().lock();
-
- try {
- // First check done concurrently, to avoid synchronization and increase throughput
- if (!paging) {
- return false;
- }
- } finally {
- lock.readLock().unlock();
- }
-
if (pageFull) {
if (message.isLargeMessage()) {
((LargeServerMessage) message).deleteFile();
@@ -1262,8 +1332,10 @@ public class PagingStoreImpl implements PagingStore {
Transaction tx,
RouteContextList listCtx,
Function<Message, Message> pageDecorator) throws Exception {
- lock.writeLock().lock();
-
+ // We need to use a readLock as we need to keep paging until we scheduled a task
+ // notice that to leave paging you need pending tasks done
+ readLock();
+ PagedMessage pagedMessage;
try {
if (!paging) {
return false;
@@ -1277,41 +1349,59 @@ public class PagingStoreImpl implements PagingStore {
message.setPaged();
- PagedMessage pagedMessage = new PagedMessageImpl(message, routeQueues(tx, listCtx), transactionID);
-
- int bytesToWrite = pagedMessage.getEncodeSize() + PageReadWriter.SIZE_RECORD;
-
- currentPageSize += bytesToWrite;
- if (currentPageSize > pageSize && currentPage.getNumberOfMessages() > 0) {
- // Make sure nothing is currently validating or using currentPage
- openNewPage();
- currentPageSize += bytesToWrite;
- }
+ pagedMessage = new PagedMessageImpl(message, routeQueues(tx, listCtx), transactionID);
+ long persistentSize = pagedMessage.getPersistentSize() > 0 ? pagedMessage.getPersistentSize() : 0;
if (tx != null && tx.isAllowPageTransaction()) {
installPageTransaction(tx, listCtx);
}
- // the apply counter will make sure we write a record on journal
- // especially on the case for non transactional sends and paging
- // doing this will give us a possibility of recovering the page counters
- long persistentSize = pagedMessage.getPersistentSize() > 0 ? pagedMessage.getPersistentSize() : 0;
- final Page page = currentPage;
- applyPageCounters(tx, page, listCtx, persistentSize);
+ // Read the comment bellow on timedWriter.addTask call
+ timedWriter.incrementTask();
- page.write(pagedMessage);
+ applyPageCounters(tx, listCtx, persistentSize);
- if (tx == null && syncNonTransactional && message.isDurable()) {
- addSyncPoint(storageManager.getContext());
- }
+ } finally {
+ readUnlock();
+ }
- if (logger.isTraceEnabled()) {
- logger.trace("Paging message {} on pageStore {} pageNr={}", pagedMessage, getStoreName(), page.getPageId());
- }
- return true;
- } finally {
- lock.writeLock().unlock();
+ // timedWriter.addTask will use a semaphore for flow control, it can hold up to a page file pending.
+ // each write will take its number of bytes away from the semaphore.
+ //
+ // However, when page cleanup is called, it will be using the page executor, and it will require a writeLock
+ // to validate if it can leave page mode.
+ //
+ // If cleanup happens while blocking for credits inside the writeLock the system would deadLock (or starve)
+ // for that reason we increment the task within the readLock (timedWriter.incrementTask()), and call this addTask away from locks
+ //
+ // timedWriter.hasPendingIO would return pending based on incrementTask, and for that reason we can still call the addTask away from the readLock.
+ //
+ // This scenario was found when running FloodServerWithAsyncSendTest smoke test.
+ timedWriter.addTask(storageManager.getContext(), pagedMessage, tx, listCtx);
+
+ return true;
+ }
+
+ void directWritePage(PagedMessage pagedMessage, boolean lineUp, boolean originalReplicated) throws Exception {
+ int bytesToWrite = pagedMessage.getEncodeSize() + PageReadWriter.SIZE_RECORD;
+
+ currentPageSize += bytesToWrite;
+ if (currentPageSize > pageSize && currentPage.getNumberOfMessages() > 0) {
+ // Make sure nothing is currently validating or using currentPage
+ openNewPage();
+ currentPageSize += bytesToWrite;
+ }
+
+ // the apply counter will make sure we write a record on journal
+ // especially on the case for non transactional sends and paging
+ // doing this will give us a possibility of recovering the page counters
+ final Page page = currentPage;
+
+ page.write(pagedMessage, lineUp, originalReplicated);
+
+ if (logger.isTraceEnabled()) {
+ logger.trace("Paging message {} on pageStore {} pageNr={}", pagedMessage, getStoreName(), page.getPageId());
}
}
@@ -1353,11 +1443,10 @@ public class PagingStoreImpl implements PagingStore {
* This is done to prevent non tx to get out of sync in case of failures
*
* @param tx
- * @param page
* @param ctx
* @throws Exception
*/
- private void applyPageCounters(Transaction tx, Page page, RouteContextList ctx, long size) throws Exception {
+ private void applyPageCounters(Transaction tx, RouteContextList ctx, long size) throws Exception {
List<org.apache.activemq.artemis.core.server.Queue> durableQueues = ctx.getDurableQueues();
List<org.apache.activemq.artemis.core.server.Queue> nonDurableQueues = ctx.getNonDurableQueues();
for (org.apache.activemq.artemis.core.server.Queue q : durableQueues) {
@@ -1413,8 +1502,18 @@ public class PagingStoreImpl implements PagingStore {
return;
}
+ @Override
+ public boolean hasPendingIO() {
+ return timedWriter.hasPendingIO();
+ }
+
+ public PageTimedWriter getPageTimedWriter() {
+ return timedWriter;
+ }
+
@Override
public void destroy() throws Exception {
+ timedWriter.stop();
// destroy has to be executed in the same executor as the cleanup
execute(this::internalDestroy);
OperationContext context = OperationContextImpl.getContext();
@@ -1427,11 +1526,7 @@ public class PagingStoreImpl implements PagingStore {
private void internalDestroy() {
try (ArtemisCloseable readLock = storageManager.closeableReadLock()) {
- while (true) {
- if (lock(100)) {
- break;
- }
- }
+ writeLock();
try {
SequentialFileFactory factory = fileFactory;
@@ -1443,7 +1538,7 @@ public class PagingStoreImpl implements PagingStore {
}
}
} finally {
- unlock();
+ writeUnlock();
try {
stop();
} catch (Exception e2) {
@@ -1498,24 +1593,11 @@ public class PagingStoreImpl implements PagingStore {
@Override
public void beforeCommit(final Transaction tx) throws Exception {
- if (!tx.isAsync()) {
- syncStore();
- }
storePageTX(tx);
}
- /**
- * @throws Exception
- */
- private void syncStore() throws Exception {
- for (PagingStore store : usedStores) {
- store.addSyncPoint(storageManager.getContext());
- }
- }
-
@Override
public void beforePrepare(final Transaction tx) throws Exception {
- syncStore();
storePageTX(tx);
}
@@ -1544,42 +1626,36 @@ public class PagingStoreImpl implements PagingStore {
}
private void openNewPage() throws Exception {
- lock.writeLock().lock();
+ numberOfPages++;
- try {
- numberOfPages++;
+ checkNumberOfPages();
- checkNumberOfPages();
+ final long newPageId = currentPageId + 1;
- final long newPageId = currentPageId + 1;
-
- if (logger.isTraceEnabled()) {
- logger.trace("destination {} new pageNr={}", storeName, newPageId);
- }
+ if (logger.isTraceEnabled()) {
+ logger.trace("destination {} new pageNr={}", storeName, newPageId);
+ }
- final Page oldPage = currentPage;
- if (oldPage != null) {
- oldPage.close(true);
- oldPage.usageDown();
- currentPage = null;
- }
+ final Page oldPage = currentPage;
+ if (oldPage != null) {
+ oldPage.close(true);
+ oldPage.usageDown();
+ currentPage = null;
+ }
- final Page newPage = newPageObject(newPageId);
+ final Page newPage = newPageObject(newPageId);
- resetCurrentPage(newPage);
+ resetCurrentPage(newPage);
- currentPageSize = 0;
+ currentPageSize = 0;
- newPage.open(true);
+ newPage.open(true);
- currentPageId = newPageId;
+ currentPageId = newPageId;
- if (newPageId < firstPageId) {
- logger.debug("open new page, setting firstPageId = {}, it was {} before", newPageId, firstPageId);
- firstPageId = newPageId;
- }
- } finally {
- lock.writeLock().unlock();
+ if (newPageId < firstPageId) {
+ logger.debug("open new page, setting firstPageId = {}, it was {} before", newPageId, firstPageId);
+ firstPageId = newPageId;
}
}
@@ -1639,7 +1715,7 @@ public class PagingStoreImpl implements PagingStore {
@Override
public Collection<Integer> getCurrentIds() throws Exception {
- lock.readLock().lock();
+ readLock();
try {
List<Integer> ids = new ArrayList<>();
SequentialFileFactory factory = fileFactory;
@@ -1650,7 +1726,7 @@ public class PagingStoreImpl implements PagingStore {
}
return ids;
} finally {
- lock.readLock().unlock();
+ readUnlock();
}
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
index d2e1746780..585c12627e 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/StorageManager.java
@@ -85,6 +85,10 @@ public interface StorageManager extends MapStorageManager, IDGenerator, ActiveMQ
return Long.MAX_VALUE;
}
+ default boolean isReplicated() {
+ return false;
+ }
+
default long getWarningRecordSize() {
/** Null journal is pretty much memory */
return Long.MAX_VALUE;
@@ -132,7 +136,7 @@ public interface StorageManager extends MapStorageManager, IDGenerator, ActiveMQ
void pageDeleted(SimpleString address, long pageNumber);
- void pageWrite(SimpleString address, PagedMessage message, long pageNumber);
+ void pageWrite(SimpleString address, PagedMessage message, long pageNumber, boolean storageUp, boolean originallyReplicated);
void afterCompleteOperations(IOCallback run);
@@ -509,6 +513,10 @@ public interface StorageManager extends MapStorageManager, IDGenerator, ActiveMQ
*/
void deleteID(long journalD) throws Exception;
+ default ArtemisCloseable closeableReadLock() {
+ return closeableReadLock(false);
+ }
+
/**
* Read lock the StorageManager. USE WITH CARE!
* <p>
@@ -518,7 +526,7 @@ public interface StorageManager extends MapStorageManager, IDGenerator, ActiveMQ
* when starting replication sync.
*
*/
- ArtemisCloseable closeableReadLock();
+ ArtemisCloseable closeableReadLock(boolean tryLock);
/**
* Closes the {@link IDGenerator} persisting the current record ID.
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
index a2d70d60ff..7790264363 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
@@ -459,7 +459,7 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
}
@Override
- public ArtemisCloseable closeableReadLock() {
+ public ArtemisCloseable closeableReadLock(boolean tryLock) {
if (reentrant.get()) {
return dummyCloseable;
}
@@ -467,7 +467,14 @@ public abstract class AbstractJournalStorageManager extends CriticalComponentImp
reentrant.set(true);
CriticalCloseable measure = measureCritical(CRITICAL_STORE);
- storageManagerLock.readLock().lock();
+
+ if (tryLock) {
+ if (!storageManagerLock.readLock().tryLock()) {
+ return null;
+ }
+ } else {
+ storageManagerLock.readLock().lock();
+ }
if (CriticalMeasure.isDummy(measure)) {
// The next statement could have been called like this:
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
index e5693b56b4..faae02ea09 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManager.java
@@ -216,10 +216,14 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
// Life Cycle Handlers
@Override
protected void beforeStart() throws Exception {
+ createDirectories();
+ cleanupIncompleteFiles();
+ }
+
+ protected void createDirectories() {
checkAndCreateDir(config.getBindingsLocation(), config.isCreateBindingsDir());
checkAndCreateDir(config.getJournalLocation(), config.isCreateJournalDir());
checkAndCreateDir(config.getLargeMessagesLocation(), config.isCreateJournalDir());
- cleanupIncompleteFiles();
}
@Override
@@ -234,6 +238,7 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
stop(false, true);
}
+ @Override
public boolean isReplicated() {
return replicator != null && replicator.isStarted();
}
@@ -394,7 +399,7 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
}
@Override
- public void pageWrite(final SimpleString address, final PagedMessage message, final long pageNumber) {
+ public void pageWrite(final SimpleString address, final PagedMessage message, final long pageNumber, boolean storageUp, boolean originallyReplicated) {
if (messageJournal.isHistory()) {
try (ArtemisCloseable lock = closeableReadLock()) {
@@ -412,16 +417,28 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
logger.warn(e.getMessage(), e);
}
}
- if (isReplicated()) {
- // Note: (https://issues.jboss.org/browse/HORNETQ-1059)
- // We have to replicate durable and non-durable messages on paging
- // since acknowledgments are written using the page-position.
- // Say you are sending durable and non-durable messages to a page
- // The ACKs would be done to wrong positions, and the backup would be a mess
+ if (isReplicated()) {
try (ArtemisCloseable lock = closeableReadLock()) {
- if (isReplicated())
- replicator.pageWrite(address, message, pageNumber);
+ if (isReplicated()) {
+ replicator.pageWrite(address, message, pageNumber, storageUp);
+ } else {
+ if (originallyReplicated) {
+ // this is a case where we originally contemplated as replicated, so we have to discount it in case it is not any more
+ OperationContext context = OperationContextImpl.getContext();
+ if (context != null) {
+ context.replicationDone();
+ }
+ }
+ }
+ }
+ } else {
+ if (originallyReplicated) {
+ // this is a case where we originally contemplated as replicated, so we have to discount it in case it is not any more
+ OperationContext context = OperationContextImpl.getContext();
+ if (context != null) {
+ context.replicationDone();
+ }
}
}
}
@@ -568,6 +585,10 @@ public class JournalStorageManager extends AbstractJournalStorageManager {
return datafiles;
}
+ void setReplicator(ReplicationManager replicator) {
+ this.replicator = replicator;
+ }
+
@Override
public void startReplication(ReplicationManager replicationManager,
PagingManager pagingManager,
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
index 2697ef53da..a2bd9563a9 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/nullpm/NullStorageManager.java
@@ -25,6 +25,7 @@ import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Consumer;
+import java.util.function.Supplier;
import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
import org.apache.activemq.artemis.api.core.ActiveMQException;
@@ -81,6 +82,8 @@ public class NullStorageManager implements StorageManager {
private volatile boolean started;
+ Supplier<OperationContext> contextSupplier = () -> dummyContext;
+
private final IOCriticalErrorListener ioCriticalErrorListener;
@Override
@@ -95,7 +98,7 @@ public class NullStorageManager implements StorageManager {
private static final ArtemisCloseable dummy = () -> { };
@Override
- public ArtemisCloseable closeableReadLock() {
+ public ArtemisCloseable closeableReadLock(boolean tryLock) {
return dummy;
}
@@ -405,7 +408,7 @@ public class NullStorageManager implements StorageManager {
}
@Override
- public void pageWrite(final SimpleString address, final PagedMessage message, final long pageNumber) {
+ public void pageWrite(final SimpleString address, final PagedMessage message, final long pageNumber, boolean lineup, boolean originallyReplicated) {
}
@Override
@@ -440,9 +443,14 @@ public class NullStorageManager implements StorageManager {
public void waitOnOperations() throws Exception {
}
+ public NullStorageManager setContextSupplier(Supplier<OperationContext> contextSupplier) {
+ this.contextSupplier = contextSupplier;
+ return this;
+ }
+
@Override
public OperationContext getContext() {
- return NullStorageManager.dummyContext;
+ return contextSupplier.get();
}
@Override
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java
index 9325ab6dc2..2e97840d4c 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/replication/ReplicationManager.java
@@ -298,9 +298,9 @@ public final class ReplicationManager implements ActiveMQComponent {
}
}
- public void pageWrite(final SimpleString address, final PagedMessage message, final long pageNumber) {
+ public void pageWrite(final SimpleString address, final PagedMessage message, final long pageNumber, boolean lineup) {
if (started) {
- sendReplicatePacket(new ReplicationPageWriteMessage(message, pageNumber, remotingConnection.isVersionUsingLongOnPageReplication(), address));
+ sendReplicatePacket(new ReplicationPageWriteMessage(message, pageNumber, remotingConnection.isVersionUsingLongOnPageReplication(), address), lineup);
}
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
index e108a4eea4..ea4d02523d 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/ActiveMQServerImpl.java
@@ -3068,9 +3068,9 @@ public class ActiveMQServerImpl implements ActiveMQServer {
protected PagingStoreFactory getPagingStoreFactory() throws Exception {
if (configuration.getStoreConfiguration() != null && configuration.getStoreConfiguration().getStoreType() == StoreConfiguration.StoreType.DATABASE) {
DatabaseStorageConfiguration dbConf = (DatabaseStorageConfiguration) configuration.getStoreConfiguration();
- return new PagingStoreFactoryDatabase(dbConf, storageManager, configuration.getPageSyncTimeout(), scheduledPool, pageExecutorFactory, ioExecutorFactory, false, ioCriticalErrorListener);
+ return new PagingStoreFactoryDatabase(dbConf, storageManager, configuration.getPageSyncTimeout(), scheduledPool, pageExecutorFactory, false, ioCriticalErrorListener);
} else {
- return new PagingStoreFactoryNIO(storageManager, configuration.getPagingLocation(), configuration.getPageSyncTimeout(), scheduledPool, pageExecutorFactory, ioExecutorFactory, configuration.isJournalSyncNonTransactional(), ioCriticalErrorListener);
+ return new PagingStoreFactoryNIO(storageManager, configuration.getPagingLocation(), configuration.getPageSyncTimeout(), scheduledPool, pageExecutorFactory, configuration.isJournalSyncNonTransactional(), ioCriticalErrorListener);
}
}
diff --git a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java
index 103057498f..3edfe8ce9f 100644
--- a/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java
+++ b/artemis-server/src/main/java/org/apache/activemq/artemis/core/server/impl/RoutingContextImpl.java
@@ -365,7 +365,7 @@ public class RoutingContextImpl implements RoutingContext {
return this.map;
}
- private static class ContextListing implements RouteContextList {
+ public static class ContextListing implements RouteContextList {
private final List<Queue> durableQueue = new ArrayList<>(1);
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java
index 37c1af0cfd..926ebf676a 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/config/impl/ConfigurationImplTest.java
@@ -102,6 +102,7 @@ import org.apache.activemq.artemis.logs.AssertionLoggerHandler;
import org.apache.activemq.artemis.json.JsonObjectBuilder;
import org.apache.activemq.artemis.utils.JsonLoader;
import org.apache.activemq.artemis.utils.RandomUtil;
+import org.apache.activemq.artemis.utils.actors.ArtemisExecutor;
import org.apache.activemq.artemis.utils.critical.CriticalAnalyzerPolicy;
import org.apache.commons.lang3.ClassUtils;
import org.junit.jupiter.api.Test;
@@ -1441,7 +1442,7 @@ public class ConfigurationImplTest extends AbstractConfigurationTestBase {
assertEquals(300000L, configuration.getAddressSettings().get("#").getPageLimitBytes().longValue());
assertEquals("DROP", configuration.getAddressSettings().get("#").getPageFullMessagePolicy().toString());
- PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), (ScheduledExecutorService) null, 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), null, null, true);
+ PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), mockScheduledExecutor(), 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), Mockito.mock(ArtemisExecutor.class), true);
assertEquals(300L, storeImpl.getPageLimitMessages().longValue());
assertEquals(300000L, storeImpl.getPageLimitBytes().longValue());
@@ -1471,7 +1472,7 @@ public class ConfigurationImplTest extends AbstractConfigurationTestBase {
assertEquals((Long)300000L, configuration.getAddressSettings().get("#").getPageLimitBytes());
assertNull(configuration.getAddressSettings().get("#").getPageFullMessagePolicy());
- PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), (ScheduledExecutorService) null, 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), null, null, true);
+ PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), mockScheduledExecutor(), 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), Mockito.mock(ArtemisExecutor.class), true);
assertNull(storeImpl.getPageLimitMessages());
assertNull(storeImpl.getPageLimitBytes());
@@ -1502,7 +1503,7 @@ public class ConfigurationImplTest extends AbstractConfigurationTestBase {
assertNull(configuration.getAddressSettings().get("#").getPageLimitBytes());
assertNull(configuration.getAddressSettings().get("#").getPageFullMessagePolicy());
- PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), (ScheduledExecutorService) null, 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), null, null, true);
+ PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), mockScheduledExecutor(), 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), Mockito.mock(ArtemisExecutor.class), true);
assertNull(storeImpl.getPageLimitMessages());
assertNull(storeImpl.getPageLimitBytes());
@@ -1532,7 +1533,7 @@ public class ConfigurationImplTest extends AbstractConfigurationTestBase {
assertEquals((Long)300000L, configuration.getAddressSettings().get("#").getPageLimitBytes());
assertNull(configuration.getAddressSettings().get("#").getPageFullMessagePolicy());
- PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), (ScheduledExecutorService) null, 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), null, null, true);
+ PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), mockScheduledExecutor(), 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), Mockito.mock(ArtemisExecutor.class), true);
assertNull(storeImpl.getPageLimitMessages());
assertNull(storeImpl.getPageLimitBytes());
@@ -1562,7 +1563,7 @@ public class ConfigurationImplTest extends AbstractConfigurationTestBase {
assertNull(configuration.getAddressSettings().get("#").getPageLimitBytes());
assertEquals("DROP", configuration.getAddressSettings().get("#").getPageFullMessagePolicy().toString());
- PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), (ScheduledExecutorService) null, 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), null, null, true);
+ PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), mockScheduledExecutor(), 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), Mockito.mock(ArtemisExecutor.class), true);
assertNull(storeImpl.getPageLimitMessages());
assertNull(storeImpl.getPageLimitBytes());
@@ -1592,7 +1593,7 @@ public class ConfigurationImplTest extends AbstractConfigurationTestBase {
assertEquals(-1L, configuration.getAddressSettings().get("#").getPageLimitBytes().longValue());
assertEquals("DROP", configuration.getAddressSettings().get("#").getPageFullMessagePolicy().toString());
- PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), (ScheduledExecutorService) null, 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), null, null, true);
+ PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), mockScheduledExecutor(), 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), Mockito.mock(ArtemisExecutor.class), true);
assertNull(storeImpl.getPageLimitMessages());
assertNull(storeImpl.getPageLimitBytes());
@@ -1618,7 +1619,7 @@ public class ConfigurationImplTest extends AbstractConfigurationTestBase {
assertEquals(333, configuration.getAddressSettings().get("#").getPrefetchPageMessages());
assertEquals(777, configuration.getAddressSettings().get("#").getPrefetchPageBytes());
- PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), (ScheduledExecutorService) null, 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), null, null, true);
+ PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), mockScheduledExecutor(), 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), Mockito.mock(ArtemisExecutor.class), true);
assertEquals(333, storeImpl.getPrefetchPageMessages());
assertEquals(777, storeImpl.getPrefetchPageBytes());
@@ -1643,7 +1644,7 @@ public class ConfigurationImplTest extends AbstractConfigurationTestBase {
assertEquals(333, configuration.getAddressSettings().get("#").getMaxReadPageMessages());
assertEquals(777, configuration.getAddressSettings().get("#").getMaxReadPageBytes());
- PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), (ScheduledExecutorService) null, 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), null, null, true);
+ PagingStore storeImpl = new PagingStoreImpl(SimpleString.of("Test"), mockScheduledExecutor(), 100L, Mockito.mock(PagingManager.class), Mockito.mock(StorageManager.class), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("Test"), configuration.getAddressSettings().get("#"), Mockito.mock(ArtemisExecutor.class), true);
assertEquals(333, storeImpl.getPrefetchPageMessages());
assertEquals(777, storeImpl.getPrefetchPageBytes());
@@ -2534,4 +2535,8 @@ public class ConfigurationImplTest extends AbstractConfigurationTestBase {
return this;
}
}
+
+ private ScheduledExecutorService mockScheduledExecutor() {
+ return Mockito.mock(ScheduledExecutorService.class);
+ }
}
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/paging/cursor/impl/ConcurrentAckTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/paging/cursor/impl/ConcurrentAckTest.java
index 81a68e6a06..dbc634bc76 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/paging/cursor/impl/ConcurrentAckTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/paging/cursor/impl/ConcurrentAckTest.java
@@ -59,7 +59,7 @@ public class ConcurrentAckTest extends ServerTestBase {
private void testConcurrentAddAckPaging(ScheduledExecutorService scheduledExecutorService, ExecutorService service) throws Throwable {
AtomicInteger errors = new AtomicInteger(0);
- PagingStoreImpl store = new PagingStoreImpl(SimpleString.of("TEST"), scheduledExecutorService, 100L, Mockito.mock(PagingManager.class), new NullStorageManager(), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("TEST"), new AddressSettings(), ArtemisExecutor.delegate(service), ArtemisExecutor.delegate(service), false);
+ PagingStoreImpl store = new PagingStoreImpl(SimpleString.of("TEST"), scheduledExecutorService, 100L, Mockito.mock(PagingManager.class), new NullStorageManager(), Mockito.mock(SequentialFileFactory.class), Mockito.mock(PagingStoreFactory.class), SimpleString.of("TEST"), new AddressSettings(), ArtemisExecutor.delegate(service), false);
PageCursorProviderImpl pageCursorProvider = new PageCursorProviderImpl(store, new NullStorageManager());
PageSubscriptionImpl subscription = (PageSubscriptionImpl) pageCursorProvider.createSubscription(1, null, true);
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/files/FileMoveManagerTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/files/FileMoveManagerTest.java
index 535c623a90..a1f62aabab 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/files/FileMoveManagerTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/server/files/FileMoveManagerTest.java
@@ -30,6 +30,7 @@ import java.io.InputStreamReader;
import java.io.PrintWriter;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import org.apache.activemq.artemis.api.config.ActiveMQDefaultConfiguration;
import org.apache.activemq.artemis.api.core.SimpleString;
@@ -50,6 +51,7 @@ import org.apache.activemq.artemis.utils.FileUtil;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
+import org.mockito.Mockito;
public class FileMoveManagerTest extends ArtemisTestCase {
@@ -291,6 +293,7 @@ public class FileMoveManagerTest extends ArtemisTestCase {
public void testMoveOverPaging() throws Exception {
ExecutorService threadPool = Executors.newCachedThreadPool();
+ runAfter(threadPool::shutdownNow);
try (AssertionLoggerHandler loggerHandler = new AssertionLoggerHandler()) {
manager.setMaxFolders(3);
for (int i = 1; i <= 10; i++) {
@@ -301,7 +304,7 @@ public class FileMoveManagerTest extends ArtemisTestCase {
final StorageManager storageManager = new NullStorageManager();
- PagingStoreFactoryNIO storeFactory = new PagingStoreFactoryNIO(storageManager, dataLocation, 100, null, new OrderedExecutorFactory(threadPool), new OrderedExecutorFactory(threadPool), true, null);
+ PagingStoreFactoryNIO storeFactory = new PagingStoreFactoryNIO(storageManager, dataLocation, 100, Mockito.mock(ScheduledExecutorService.class), new OrderedExecutorFactory(threadPool), true, null);
PagingManagerImpl managerImpl = new PagingManagerImpl(storeFactory, addressSettings, -1, -1, ActiveMQDefaultConfiguration.getDefaultManagementAddress(), null);
@@ -321,8 +324,6 @@ public class FileMoveManagerTest extends ArtemisTestCase {
}
assertFalse(loggerHandler.findText("address.txt"), "The loggers are complaining about address.txt");
- } finally {
- threadPool.shutdown();
}
}
diff --git a/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java b/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java
index 83ad4d8db7..6a6b005526 100644
--- a/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java
+++ b/artemis-server/src/test/java/org/apache/activemq/artemis/core/transaction/impl/TransactionImplTest.java
@@ -348,7 +348,7 @@ public class TransactionImplTest extends ServerTestBase {
}
@Override
- public ArtemisCloseable closeableReadLock() {
+ public ArtemisCloseable closeableReadLock(boolean tryLock) {
return () -> { };
}
@@ -423,7 +423,7 @@ public class TransactionImplTest extends ServerTestBase {
}
@Override
- public void pageWrite(SimpleString address, PagedMessage message, long pageNumber) {
+ public void pageWrite(SimpleString address, PagedMessage message, long pageNumber, boolean lineup, boolean originallyReplicated) {
}
diff --git a/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderTestAccessor.java b/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderTestAccessor.java
index 3e3b541749..1059c7b9aa 100644
--- a/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderTestAccessor.java
+++ b/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageCursorProviderTestAccessor.java
@@ -17,12 +17,27 @@
package org.apache.activemq.artemis.core.paging.cursor.impl;
+import java.util.concurrent.TimeUnit;
+
import org.apache.activemq.artemis.core.paging.cursor.PageCursorProvider;
+import org.apache.activemq.artemis.utils.SimpleFutureImpl;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
public class PageCursorProviderTestAccessor {
public static void cleanup(PageCursorProvider provider) {
- ((PageCursorProviderImpl)provider).cleanup();
+ SimpleFutureImpl<Object> simpleFuture = new SimpleFutureImpl<>();
+ ((PageCursorProviderImpl)provider).pagingStore.execute(() -> {
+ ((PageCursorProviderImpl)provider).cleanup();
+ simpleFuture.set(new Object());
+ });
+ try {
+ Object value = simpleFuture.get(30, TimeUnit.SECONDS);
+ assertNotNull(value, "There was a timeout on calling cleanup");
+ } catch (Throwable e) {
+ e.printStackTrace();
+ }
}
public static Long getNumberOfMessagesOnSubscriptions(PageCursorProvider provider) {
diff --git a/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java b/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java
index 49a681df01..5d319708ef 100644
--- a/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java
+++ b/tests/artemis-test-support/src/main/java/org/apache/activemq/artemis/tests/util/ActiveMQTestBase.java
@@ -58,7 +58,9 @@ import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.UUID;
+import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
@@ -94,6 +96,7 @@ import org.apache.activemq.artemis.core.config.StoreConfiguration;
import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
import org.apache.activemq.artemis.core.config.impl.SecurityConfiguration;
import org.apache.activemq.artemis.core.config.storage.DatabaseStorageConfiguration;
+import org.apache.activemq.artemis.core.io.IOCallback;
import org.apache.activemq.artemis.core.io.SequentialFileFactory;
import org.apache.activemq.artemis.core.io.aio.AIOSequentialFileFactory;
import org.apache.activemq.artemis.core.io.nio.NIOSequentialFileFactory;
@@ -104,6 +107,7 @@ import org.apache.activemq.artemis.core.journal.impl.JournalImpl;
import org.apache.activemq.artemis.core.journal.impl.JournalReaderCallback;
import org.apache.activemq.artemis.core.message.impl.CoreMessage;
import org.apache.activemq.artemis.core.paging.PagingStore;
+import org.apache.activemq.artemis.core.persistence.OperationContext;
import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
import org.apache.activemq.artemis.core.postoffice.Binding;
import org.apache.activemq.artemis.core.postoffice.Bindings;
@@ -152,6 +156,7 @@ import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
import org.apache.activemq.artemis.utils.Env;
import org.apache.activemq.artemis.utils.FileUtil;
import org.apache.activemq.artemis.utils.RandomUtil;
+import org.apache.activemq.artemis.utils.SimpleFutureImpl;
import org.apache.activemq.artemis.utils.ThreadDumpUtil;
import org.apache.activemq.artemis.utils.UUIDGenerator;
import org.apache.activemq.artemis.utils.Wait;
@@ -458,7 +463,7 @@ public abstract class ActiveMQTestBase extends ArtemisTestCase {
}
protected ConfigurationImpl createBasicConfig(final int serverID) {
- ConfigurationImpl configuration = new ConfigurationImpl().setSecurityEnabled(false).setJournalMinFiles(2).setJournalFileSize(100 * 1024).setJournalType(getDefaultJournalType()).setJournalDirectory(getJournalDir(serverID, false)).setBindingsDirectory(getBindingsDir(serverID, false)).setPagingDirectory(getPageDir(serverID, false)).setLargeMessagesDirectory(getLargeMessagesDir(serverID, false)).setJournalCompactMinFiles(0).setJournalCompactPercentage(0).setClusterPassword(CLUSTER_PASS [...]
+ ConfigurationImpl configuration = new ConfigurationImpl().setSecurityEnabled(false).setJournalMinFiles(2).setJournalFileSize(100 * 1024).setJournalType(getDefaultJournalType()).setJournalDirectory(getJournalDir(serverID, false)).setBindingsDirectory(getBindingsDir(serverID, false)).setPagingDirectory(getPageDir(serverID, false)).setLargeMessagesDirectory(getLargeMessagesDir(serverID, false)).setJournalCompactMinFiles(0).setJournalCompactPercentage(0).setClusterPassword(CLUSTER_PASS [...]
// When it comes to the testsuite, we don't need any batching, I will leave some minimal batching to exercise the codebase
configuration.setJournalBufferTimeout_AIO(100).setJournalBufferTimeout_NIO(100);
@@ -2597,4 +2602,52 @@ public abstract class ActiveMQTestBase extends ArtemisTestCase {
public static void waitForLatch(CountDownLatch latch) throws InterruptedException {
assertTrue(latch.await(1, TimeUnit.MINUTES), "Latch has got to return within a minute");
}
+
+ /** use the current OperationContext to sync any pending operations. */
+ protected void syncOperationContext() throws Exception {
+ CountDownLatch latch = new CountDownLatch(1);
+ OperationContext operationContext = OperationContextImpl.getContext();
+ if (operationContext != null) {
+ operationContext.executeOnCompletion(new IOCallback() {
+ @Override
+ public void done() {
+ latch.countDown();
+ }
+
+ @Override
+ public void onError(int errorCode, String errorMessage) {
+
+ }
+ });
+
+ assertTrue(latch.await(10, TimeUnit.SECONDS));
+ }
+ }
+
+ protected <T> T callOnExecutor(Executor executor, Callable<T> callable, long time, TimeUnit unit) throws Exception {
+ Object nullReturn = new Object();
+ SimpleFutureImpl<Object> future = new SimpleFutureImpl<>();
+ executor.execute(() -> {
+ try {
+ Object value = callable.call();
+ if (value == null) {
+ future.set(nullReturn);
+ } else {
+ future.set(value);
+ }
+ } catch (Exception e) {
+ future.fail(e);
+ }
+ });
+
+ Object futureGet = future.get(time, unit);
+
+ assertNotNull(futureGet, "Timeout call");
+
+ if (futureGet == nullReturn) {
+ return null;
+ } else {
+ return (T) futureGet;
+ }
+ }
}
diff --git a/tests/db-tests/src/test/java/org/apache/activemq/artemis/tests/db/paging/PageSizeTest.java b/tests/db-tests/src/test/java/org/apache/activemq/artemis/tests/db/paging/PageSizeTest.java
index b58209f718..9ded7b5615 100644
--- a/tests/db-tests/src/test/java/org/apache/activemq/artemis/tests/db/paging/PageSizeTest.java
+++ b/tests/db-tests/src/test/java/org/apache/activemq/artemis/tests/db/paging/PageSizeTest.java
@@ -106,8 +106,7 @@ public class PageSizeTest extends ParameterDBTestBase {
// Since we sent a large message (I mean, not streaming large, but larger than page-size,
// a page with more than 30K should been created to allow the "large" message to be stored.
- size = getMaxSizeBytesStored(queue);
- assertTrue(size >= 50 * 1024, "size is " + size);
+ Wait.assertTrue(() -> getMaxSizeBytesStored(queue) >= 50 * 1024);
}
}
diff --git a/tests/db-tests/src/test/java/org/apache/activemq/artemis/tests/db/paging/PagingTest.java b/tests/db-tests/src/test/java/org/apache/activemq/artemis/tests/db/paging/PagingTest.java
index 9df22b80d6..916766e851 100644
--- a/tests/db-tests/src/test/java/org/apache/activemq/artemis/tests/db/paging/PagingTest.java
+++ b/tests/db-tests/src/test/java/org/apache/activemq/artemis/tests/db/paging/PagingTest.java
@@ -105,7 +105,6 @@ import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.persistence.impl.journal.AckDescribe;
import org.apache.activemq.artemis.core.persistence.impl.journal.DescribeJournal;
import org.apache.activemq.artemis.core.persistence.impl.journal.JournalRecordIds;
-import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
import org.apache.activemq.artemis.core.server.ActiveMQServer;
import org.apache.activemq.artemis.core.server.Queue;
import org.apache.activemq.artemis.core.server.impl.ActiveMQServerImpl;
@@ -282,7 +281,7 @@ public class PagingTest extends ParameterDBTestBase {
session.createQueue(QueueConfiguration.of(ADDRESS.concat("-0")).setAddress(ADDRESS));
- server.getPagingManager().getPageStore(ADDRESS).forceAnotherPage();
+ server.getPagingManager().getPageStore(ADDRESS).forceAnotherPage(true);
server.getPagingManager().getPageStore(ADDRESS).disableCleanup();
session.start();
@@ -616,7 +615,7 @@ public class PagingTest extends ParameterDBTestBase {
for (int j = 0; j < 10; j++) {
producer.send(session.createMessage());
}
- serverQueue.getPagingStore().forceAnotherPage();
+ serverQueue.getPagingStore().forceAnotherPage(true);
}
}
@@ -771,7 +770,7 @@ public class PagingTest extends ParameterDBTestBase {
int page = 1;
for (int i = 0; i < numberOfMessages; i++) {
if (i % 10 == 0 && i > 0) {
- queue.getPagingStore().forceAnotherPage();
+ queue.getPagingStore().forceAnotherPage(true);
page++;
}
message = session.createMessage(true);
@@ -852,7 +851,7 @@ public class PagingTest extends ParameterDBTestBase {
for (int i = 0; i < numberOfMessages; i++) {
if (i % 10 == 0 && i > 0) {
- queue.getPagingStore().forceAnotherPage();
+ queue.getPagingStore().forceAnotherPage(true);
page++;
}
message = session.createMessage(true);
@@ -983,7 +982,7 @@ public class PagingTest extends ParameterDBTestBase {
for (int i = 0; i < numberOfMessages; i++) {
if (i % 10 == 0 && i > 0) {
- queue.getPagingStore().forceAnotherPage();
+ queue.getPagingStore().forceAnotherPage(true);
// forceAnotherPage could be called concurrently with cleanup on this case
// so we have to call startPaging to make sure we are still paging on this test
queue.getPagingStore().startPaging();
@@ -1226,7 +1225,7 @@ public class PagingTest extends ParameterDBTestBase {
if ((i + 1) % 5 == 0) {
session.commit();
- queue.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue.getPageSubscription().getPagingStore().forceAnotherPage(true);
}
}
@@ -1331,7 +1330,7 @@ public class PagingTest extends ParameterDBTestBase {
if ((i + 1) % 5 == 0) {
session.commit();
- queue.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue.getPageSubscription().getPagingStore().forceAnotherPage(true);
}
}
@@ -1414,7 +1413,7 @@ public class PagingTest extends ParameterDBTestBase {
if ((i + 1) % 5 == 0) {
session.commit();
- queue.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue.getPageSubscription().getPagingStore().forceAnotherPage(true);
}
}
@@ -1508,7 +1507,7 @@ public class PagingTest extends ParameterDBTestBase {
if ((i + 1) % 5 == 0) {
session.commit();
- queue.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue.getPageSubscription().getPagingStore().forceAnotherPage(true);
}
}
@@ -1698,9 +1697,9 @@ public class PagingTest extends ParameterDBTestBase {
@TestTemplate
public void testDeleteQueueRestart() throws Exception {
- Configuration config = createDefaultInVMConfig().setJournalDirectory(getJournalDir()).setJournalSyncNonTransactional(false).setJournalCompactMinFiles(0); // disable compact
+ Configuration config = createDefaultInVMConfig().setJournalDirectory(getJournalDir()).setJournalSyncNonTransactional(false).setJournalCompactMinFiles(0).setPageSyncTimeout(1000); // disable compact
- server = createServer(true, config, PagingTest.PAGE_SIZE, PagingTest.PAGE_MAX);
+ server = createServer(true, config, PagingTest.PAGE_SIZE, PagingTest.PAGE_MAX, -1, -1);
server.start();
@@ -1867,7 +1866,7 @@ public class PagingTest extends ParameterDBTestBase {
}
private void waitBuffer(ClientConsumerInternal clientBuffer, int bufferSize) {
- Wait.assertTrue(() -> "expected " + bufferSize + " but got " + clientBuffer.getBufferSize(), () -> clientBuffer.getBufferSize() > bufferSize, 5000, 100);
+ Wait.assertTrue(() -> "expected " + bufferSize + " but got " + clientBuffer.getBufferSize(), () -> clientBuffer.getBufferSize() > bufferSize, 15_000, 100);
}
@TestTemplate
@@ -3639,7 +3638,7 @@ public class PagingTest extends ParameterDBTestBase {
producer.send(message);
session.commit();
- queue.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue.getPageSubscription().getPagingStore().forceAnotherPage(true);
}
@@ -3706,7 +3705,7 @@ public class PagingTest extends ParameterDBTestBase {
super.cleanup();
} else {
try {
- pagingStore.unlock();
+ pagingStore.writeUnlock();
} catch (Throwable ignored) {
}
}
@@ -3716,7 +3715,7 @@ public class PagingTest extends ParameterDBTestBase {
server = new ActiveMQServerImpl(config, ManagementFactory.getPlatformMBeanServer(), new ActiveMQSecurityManagerImpl()) {
@Override
protected PagingStoreFactoryNIO getPagingStoreFactory() {
- return new PagingStoreFactoryNIO(this.getStorageManager(), this.getConfiguration().getPagingLocation(), this.getConfiguration().getJournalBufferTimeout_NIO(), this.getScheduledPool(), this.getExecutorFactory(), this.getExecutorFactory(), this.getConfiguration().isJournalSyncNonTransactional(), null) {
+ return new PagingStoreFactoryNIO(this.getStorageManager(), this.getConfiguration().getPagingLocation(), this.getConfiguration().getJournalBufferTimeout_NIO(), this.getScheduledPool(), this.getExecutorFactory(), this.getConfiguration().isJournalSyncNonTransactional(), null) {
@Override
public PageCursorProvider newCursorProvider(PagingStore store,
StorageManager storageManager,
@@ -3763,7 +3762,7 @@ public class PagingTest extends ParameterDBTestBase {
producer.send(message);
session.commit();
if (i < 19) {
- queue1.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue1.getPageSubscription().getPagingStore().forceAnotherPage(true);
}
}
@@ -3810,7 +3809,7 @@ public class PagingTest extends ParameterDBTestBase {
session.commit();
if (i == 5) {
- queue.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue.getPageSubscription().getPagingStore().forceAnotherPage(true);
}
}
@@ -4211,74 +4210,6 @@ public class PagingTest extends ParameterDBTestBase {
}
}
- @TestTemplate
- public void testSyncPage() throws Exception {
- Configuration config = createDefaultInVMConfig();
-
- server = createServer(true, config, PagingTest.PAGE_SIZE, PagingTest.PAGE_MAX);
-
- server.start();
-
- try {
- server.addAddressInfo(new AddressInfo(PagingTest.ADDRESS, RoutingType.ANYCAST));
- server.createQueue(QueueConfiguration.of(PagingTest.ADDRESS).setRoutingType(RoutingType.ANYCAST));
-
- final CountDownLatch pageUp = new CountDownLatch(0);
- final CountDownLatch pageDone = new CountDownLatch(1);
-
- OperationContext ctx = new DummyOperationContext(pageUp, pageDone);
-
- OperationContextImpl.setContext(ctx);
-
- PagingManager paging = server.getPagingManager();
-
- PagingStore store = paging.getPageStore(ADDRESS);
-
- store.addSyncPoint(OperationContextImpl.getContext());
-
- assertTrue(pageUp.await(10, TimeUnit.SECONDS));
-
- assertTrue(pageDone.await(10, TimeUnit.SECONDS));
-
- server.stop();
-
- } finally {
- try {
- server.stop();
- } catch (Throwable ignored) {
- }
-
- OperationContextImpl.clearContext();
- }
-
- }
-
- @TestTemplate
- public void testSyncPageTX() throws Exception {
- Configuration config = createDefaultInVMConfig();
-
- server = createServer(true, config, PagingTest.PAGE_SIZE, PagingTest.PAGE_MAX);
-
- server.start();
- server.createQueue(QueueConfiguration.of(PagingTest.ADDRESS).setRoutingType(RoutingType.ANYCAST));
-
- final CountDownLatch pageUp = new CountDownLatch(0);
- final CountDownLatch pageDone = new CountDownLatch(1);
-
- OperationContext ctx = new DummyOperationContext(pageUp, pageDone);
- OperationContextImpl.setContext(ctx);
-
- PagingManager paging = server.getPagingManager();
-
- PagingStore store = paging.getPageStore(ADDRESS);
-
- store.addSyncPoint(OperationContextImpl.getContext());
-
- assertTrue(pageUp.await(10, TimeUnit.SECONDS));
-
- assertTrue(pageDone.await(10, TimeUnit.SECONDS));
- }
-
@TestTemplate
public void testPagingOneDestinationOnly() throws Exception {
SimpleString PAGED_ADDRESS = SimpleString.of("paged");
@@ -5380,7 +5311,7 @@ public class PagingTest extends ParameterDBTestBase {
msg = cons2.receive(5000);
assertNotNull(msg);
- queue.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue.getPageSubscription().getPagingStore().forceAnotherPage(true);
msg = session.createMessage(true);
msg.putIntProperty("dest", 1);
@@ -5462,7 +5393,7 @@ public class PagingTest extends ParameterDBTestBase {
session.commit();
- pageStore.forceAnotherPage();
+ pageStore.forceAnotherPage(true);
for (int i = 0; i < messagesSentAfterBurst; i++) {
{
@@ -5488,7 +5419,7 @@ public class PagingTest extends ParameterDBTestBase {
if (i > 0 && i % 10 == 0) {
session.commit();
if (i + 10 < messagesSentAfterBurst) {
- pageStore.forceAnotherPage();
+ pageStore.forceAnotherPage(true);
}
}
}
@@ -5596,7 +5527,7 @@ public class PagingTest extends ParameterDBTestBase {
prod.send(msg);
if (i > 0 && i % 10 == 0) {
- store.forceAnotherPage();
+ store.forceAnotherPage(true);
}
}
@@ -5686,7 +5617,7 @@ public class PagingTest extends ParameterDBTestBase {
if (i > 0 && i % 10 == 0) {
store.startPaging();
- store.forceAnotherPage();
+ store.forceAnotherPage(true);
}
}
@@ -5698,7 +5629,7 @@ public class PagingTest extends ParameterDBTestBase {
if (i > 0 && i % 10 == 0) {
store.startPaging();
- store.forceAnotherPage();
+ store.forceAnotherPage(true);
}
}
@@ -5764,7 +5695,7 @@ public class PagingTest extends ParameterDBTestBase {
prod.send(msg);
if ((i + 1) % 5 == 0 && i < 50) {
session.commit();
- store.forceAnotherPage();
+ store.forceAnotherPage(true);
}
}
session.commit();
@@ -5861,7 +5792,7 @@ public class PagingTest extends ParameterDBTestBase {
ClientMessage msgReceivedCons2 = cons2.receive(1000);
assertNotNull(msgReceivedCons2);
- store.forceAnotherPage();
+ store.forceAnotherPage(true);
msg = session.createMessage(true);
@@ -5969,7 +5900,7 @@ public class PagingTest extends ParameterDBTestBase {
prod.send(msg);
if (i % 10 == 0) {
session.commit();
- store.forceAnotherPage();
+ store.forceAnotherPage(true);
}
}
session.commit();
@@ -6042,7 +5973,7 @@ public class PagingTest extends ParameterDBTestBase {
AddressSettings addressSettings,
ArtemisExecutor executor,
boolean syncNonTransactional) {
- super(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, fileFactory, storeFactory, storeName, addressSettings, executor, executor, syncNonTransactional);
+ super(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, fileFactory, storeFactory, storeName, addressSettings, executor, syncNonTransactional);
}
/**
@@ -6059,7 +5990,7 @@ public class PagingTest extends ParameterDBTestBase {
server = new ActiveMQServerImpl(config, ManagementFactory.getPlatformMBeanServer(), new ActiveMQSecurityManagerImpl()) {
@Override
protected PagingStoreFactoryDatabase getPagingStoreFactory() throws Exception {
- return new PagingStoreFactoryDatabase((DatabaseStorageConfiguration) this.getConfiguration().getStoreConfiguration(), this.getStorageManager(), this.getConfiguration().getJournalBufferTimeout_NIO(), this.getScheduledPool(), this.getExecutorFactory(), this.getExecutorFactory(), this.getConfiguration().isJournalSyncNonTransactional(), null) {
+ return new PagingStoreFactoryDatabase((DatabaseStorageConfiguration) this.getConfiguration().getStoreConfiguration(), this.getStorageManager(), this.getConfiguration().getJournalBufferTimeout_NIO(), this.getScheduledPool(), this.getExecutorFactory(), this.getConfiguration().isJournalSyncNonTransactional(), null) {
@Override
public synchronized PagingStore newStore(SimpleString address, AddressSettings settings) {
return new NonStoppablePagingStoreImpl(address, this.getScheduledExecutor(), config.getJournalBufferTimeout_NIO(), getPagingManager(), getStorageManager(), null, this, address, settings, getExecutorFactory().getExecutor(), this.syncNonTransactional);
@@ -6071,7 +6002,7 @@ public class PagingTest extends ParameterDBTestBase {
server = new ActiveMQServerImpl(config, ManagementFactory.getPlatformMBeanServer(), new ActiveMQSecurityManagerImpl()) {
@Override
protected PagingStoreFactoryNIO getPagingStoreFactory() {
- return new PagingStoreFactoryNIO(this.getStorageManager(), this.getConfiguration().getPagingLocation(), this.getConfiguration().getJournalBufferTimeout_NIO(), this.getScheduledPool(), this.getExecutorFactory(), this.getExecutorFactory(), this.getConfiguration().isJournalSyncNonTransactional(), null) {
+ return new PagingStoreFactoryNIO(this.getStorageManager(), this.getConfiguration().getPagingLocation(), this.getConfiguration().getJournalBufferTimeout_NIO(), this.getScheduledPool(), this.getExecutorFactory(), this.getConfiguration().isJournalSyncNonTransactional(), null) {
@Override
public synchronized PagingStore newStore(SimpleString address, AddressSettings settings) {
return new NonStoppablePagingStoreImpl(address, this.getScheduledExecutor(), config.getJournalBufferTimeout_NIO(), getPagingManager(), getStorageManager(), null, this, address, settings, getExecutorFactory().getExecutor(), this.isSyncNonTransactional());
@@ -6276,7 +6207,7 @@ public class PagingTest extends ParameterDBTestBase {
session.commit();
if (forceAnotherPage) {
- queue.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue.getPageSubscription().getPagingStore().forceAnotherPage(true);
}
message = session.createMessage(persistentMessages);
@@ -6487,7 +6418,7 @@ public class PagingTest extends ParameterDBTestBase {
ClientConsumer consumer = session.createConsumer(PagingTest.ADDRESS);
session.start();
assertNull(consumer.receiveImmediate());
- assertTrue(server.getPagingManager().getTransactions().isEmpty());
+ Wait.assertTrue(() -> server.getPagingManager().getTransactions().isEmpty(), 5000);
} else {
ClientConsumer consumer = session.createConsumer(PagingTest.ADDRESS);
session.start();
@@ -6497,7 +6428,7 @@ public class PagingTest extends ParameterDBTestBase {
PageTransactionInfo pageTransactionInfo = server.getPagingManager().getTransactions().values().iterator().next();
session.rollback();
Wait.assertTrue(() -> pageTransactionInfo.isRollback(), 1000, 100);
- assertTrue(server.getPagingManager().getTransactions().isEmpty());
+ Wait.assertTrue(() -> server.getPagingManager().getTransactions().isEmpty(), 5000);
}
session.close();
diff --git a/tests/db-tests/src/test/java/org/apache/activemq/artemis/tests/db/paging/RealServerDatabasePagingTest.java b/tests/db-tests/src/test/java/org/apache/activemq/artemis/tests/db/paging/RealServerDatabasePagingTest.java
index c06f286596..6ff002bb1d 100644
--- a/tests/db-tests/src/test/java/org/apache/activemq/artemis/tests/db/paging/RealServerDatabasePagingTest.java
+++ b/tests/db-tests/src/test/java/org/apache/activemq/artemis/tests/db/paging/RealServerDatabasePagingTest.java
@@ -117,8 +117,7 @@ public class RealServerDatabasePagingTest extends ParameterDBTestBase {
}
- serverProcess.destroyForcibly();
- serverProcess.waitFor(1, TimeUnit.MINUTES);
+ stopServerWithFile(getServerLocation(database.getName()), serverProcess, 1, TimeUnit.MINUTES);
assertFalse(serverProcess.isAlive());
serverProcess = startServer(database.getName(), 0, 60_000);
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/paging/AmqpFilterChangePageTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/paging/AmqpFilterChangePageTest.java
index 117fadda10..46b1b92686 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/paging/AmqpFilterChangePageTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/paging/AmqpFilterChangePageTest.java
@@ -74,7 +74,7 @@ public class AmqpFilterChangePageTest extends ActiveMQTestBase {
producer.send(message);
if (i % 100 == 0 && i > 0) {
session.commit();
- queue.getPagingStore().forceAnotherPage();
+ queue.getPagingStore().forceAnotherPage(true);
}
}
session.commit();
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/SendAckFailTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/SendAckFailTest.java
index e609bf85a9..637f62916a 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/SendAckFailTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/SendAckFailTest.java
@@ -381,8 +381,8 @@ public class SendAckFailTest extends SpawnedTestBase {
}
@Override
- public void pageWrite(SimpleString address, PagedMessage message, long pageNumber) {
- manager.pageWrite(address, message, pageNumber);
+ public void pageWrite(SimpleString address, PagedMessage message, long pageNumber, boolean lineup, boolean originallyReplicated) {
+ manager.pageWrite(address, message, pageNumber, lineup, originallyReplicated);
}
@Override
@@ -914,7 +914,7 @@ public class SendAckFailTest extends SpawnedTestBase {
}
@Override
- public ArtemisCloseable closeableReadLock() {
+ public ArtemisCloseable closeableReadLock(boolean tryLock) {
return manager.closeableReadLock();
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/BackupSyncJournalTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/BackupSyncJournalTest.java
index 99941b2ff4..0ae372fb78 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/BackupSyncJournalTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/BackupSyncJournalTest.java
@@ -111,7 +111,7 @@ public class BackupSyncJournalTest extends FailoverTestBase {
// in case of paging I must close the current page otherwise we will get a pending counter
// what would make the verification on similar journal to fail after the recovery
if (store.isPaging()) {
- store.forceAnotherPage();
+ store.forceAnotherPage(true);
}
backupServer.start();
@@ -124,7 +124,7 @@ public class BackupSyncJournalTest extends FailoverTestBase {
// in case of paging I must close the current page otherwise we will get a pending counter
// what would make the verification on similar journal to fail after the recovery
if (store.isPaging()) {
- store.forceAnotherPage();
+ store.forceAnotherPage(true);
}
// Deliver messages with Backup up-to-date
@@ -136,7 +136,7 @@ public class BackupSyncJournalTest extends FailoverTestBase {
// in case of paging I must close the current page otherwise we will get a pending counter
// what would make the verification on similar journal to fail after the recovery
if (store.isPaging()) {
- store.forceAnotherPage();
+ store.forceAnotherPage(true);
}
int size = messageJournal.getFileSize();
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/NettyReplicatedFailoverTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/NettyReplicatedFailoverTest.java
index 55be002c7f..05c833c680 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/NettyReplicatedFailoverTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/NettyReplicatedFailoverTest.java
@@ -163,7 +163,7 @@ public class NettyReplicatedFailoverTest extends NettyFailoverInVMTest {
for (int i = 0; i < 50; i++) {
producer.send(session.createTextMessage("hello"));
session.commit();
- serverQueue.getPagingStore().forceAnotherPage();
+ serverQueue.getPagingStore().forceAnotherPage(true);
}
backupServer.stop();
backupServer.start();
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/PrimaryCrashOnBackupSyncTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/PrimaryCrashOnBackupSyncTest.java
index 742ac6a411..eae40f7c87 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/PrimaryCrashOnBackupSyncTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/PrimaryCrashOnBackupSyncTest.java
@@ -215,7 +215,7 @@ public class PrimaryCrashOnBackupSyncTest extends ActiveMQTestBase {
ActiveMQServer primaryServer = new ActiveMQServerImpl(primaryConfiguration, ManagementFactory.getPlatformMBeanServer(), new ActiveMQJAASSecurityManager(InVMLoginModule.class.getName(), new SecurityConfiguration())) {
@Override
protected PagingStoreFactoryNIO getPagingStoreFactory() {
- return new PagingStoreFactoryNIO(this.getStorageManager(), this.getConfiguration().getPagingLocation(), this.getConfiguration().getJournalBufferTimeout_NIO(), this.getScheduledPool(), this.getExecutorFactory(), this.getExecutorFactory(), this.getConfiguration().isJournalSyncNonTransactional(), null) {
+ return new PagingStoreFactoryNIO(this.getStorageManager(), this.getConfiguration().getPagingLocation(), this.getConfiguration().getJournalBufferTimeout_NIO(), this.getScheduledPool(), this.getExecutorFactory(), this.getConfiguration().isJournalSyncNonTransactional(), null) {
@Override
public synchronized PagingStore newStore(SimpleString address, AddressSettings settings) {
return new DelayPagingStoreImpl(address, this.getScheduledExecutor(), primaryConfiguration.getJournalBufferTimeout_NIO(), getPagingManager(), getStorageManager(), null, this, address, settings, getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), this.isSyncNonTransactional());
@@ -268,7 +268,7 @@ class DelayPagingStoreImpl extends PagingStoreImpl {
AddressSettings addressSettings,
ArtemisExecutor executor, ArtemisExecutor ioExecutor,
boolean syncNonTransactional) {
- super(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, fileFactory, storeFactory, storeName, addressSettings, executor, ioExecutor, syncNonTransactional);
+ super(address, scheduledExecutor, syncTimeout, pagingManager, storageManager, fileFactory, storeFactory, storeName, addressSettings, executor, syncNonTransactional);
}
@Override
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/ReplicatedPagedFailoverTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/ReplicatedPagedFailoverTest.java
index 5ea8369714..9e4541ca7b 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/ReplicatedPagedFailoverTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/ReplicatedPagedFailoverTest.java
@@ -85,7 +85,7 @@ public class ReplicatedPagedFailoverTest extends ReplicatedFailoverTest {
// some are durable, some are not!
producer.send(createMessage(session, i, i % 2 == 0));
if (i > 0 && i % messagesPerPage == 0) {
- queue.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue.getPageSubscription().getPagingStore().forceAnotherPage(true);
}
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/cluster/TopicClusterPageStoreSizeTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/cluster/TopicClusterPageStoreSizeTest.java
index 69b5969d22..4b2252b0c5 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/cluster/TopicClusterPageStoreSizeTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/jms/cluster/TopicClusterPageStoreSizeTest.java
@@ -30,12 +30,23 @@ import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;
+import java.lang.invoke.MethodHandles;
+
import org.apache.activemq.artemis.api.core.SimpleString;
import org.apache.activemq.artemis.tests.util.JMSClusteredTestBase;
+import org.apache.activemq.artemis.tests.util.Wait;
import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class TopicClusterPageStoreSizeTest extends JMSClusteredTestBase {
+ @Override
+ protected boolean enablePersistence() {
+ return true;
+ }
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
public static final String TOPIC = "jms.t1";
@Test
@@ -81,11 +92,11 @@ public class TopicClusterPageStoreSizeTest extends JMSClusteredTestBase {
if (forcePaging) {
for (SimpleString psName : server1.getPagingManager().getStoreNames()) {
- System.err.println("server1: force paging on:" + psName);
+ logger.info("server1: force paging on:{}", psName);
server1.getPagingManager().getPageStore(psName).startPaging();
}
for (SimpleString psName : server2.getPagingManager().getStoreNames()) {
- System.err.println("server2: force paging on:" + psName);
+ logger.info("server2: force paging on:{}", psName);
server2.getPagingManager().getPageStore(psName).startPaging();
}
}
@@ -95,13 +106,13 @@ public class TopicClusterPageStoreSizeTest extends JMSClusteredTestBase {
TextMessage m2 = (TextMessage) cons2.receive(5000);
assertNotNull(m2);
assertTrue(m2.getJMSDestination().toString().contains(TOPIC));
- System.err.println("sub2 on 2, no ack, message txt:" + m2.getText());
+ logger.info("sub2 on 2, no ack, message txt:{}", m2.getText());
TextMessage m1 = (TextMessage) cons1.receive(5000);
assertNotNull(m1);
assertTrue(m1.getJMSDestination().toString().contains(TOPIC));
- System.err.println("message txt:" + m1.getText());
+ logger.info("message txt: {}", m1.getText());
m1.acknowledge();
// leave m2 for reconnect on server1
@@ -110,19 +121,19 @@ public class TopicClusterPageStoreSizeTest extends JMSClusteredTestBase {
conn2.close();
for (SimpleString psName : server1.getPagingManager().getStoreNames()) {
- assertTrue(server1.getPagingManager().getPageStore(psName).getAddressSize() >= 0, "non negative size: " + psName);
+ Wait.assertTrue(() -> server1.getPagingManager().getPageStore(psName).getAddressSize() >= 0, 5000);
}
for (SimpleString psName : server2.getPagingManager().getStoreNames()) {
- System.err.println("server2: size of pages store: " + psName + " :" + server2.getPagingManager().getPageStore(psName).getAddressSize());
- assertTrue(server2.getPagingManager().getPageStore(psName).getAddressSize() >= 0, "non negative size: " + psName);
+ logger.info("server2: size of pages store: {} : {}", psName, server2.getPagingManager().getPageStore(psName).getAddressSize());
+ Wait.assertTrue(() -> server2.getPagingManager().getPageStore(psName).getAddressSize() >= 0, 5000);
}
if (forcePaging) {
// message in the store, should have getPagedSize or is there some such thing?
- assertTrue(server2.getPagingManager().getPageStore(SimpleString.of(TOPIC)).getNumberOfPages() > 0, "size on 2");
+ Wait.assertTrue(() -> server2.getPagingManager().getPageStore(SimpleString.of(TOPIC)).getNumberOfPages() > 0);
} else {
- assertTrue(server2.getPagingManager().getPageStore(SimpleString.of(TOPIC)).getAddressSize() > 0, "size on 2");
+ Wait.assertTrue(() -> server2.getPagingManager().getPageStore(SimpleString.of(TOPIC)).getAddressSize() > 0);
}
// reconnect
@@ -140,7 +151,7 @@ public class TopicClusterPageStoreSizeTest extends JMSClusteredTestBase {
m2 = (TextMessage) cons1.receive(5000);
assertNotNull(m2);
assertTrue(m2.getJMSDestination().toString().contains(TOPIC));
- System.err.println("sub2 on 1, message txt:" + m2.getText());
+ logger.info("sub2 on 1, message txt: {}", m2.getText());
m2.acknowledge();
@@ -174,7 +185,7 @@ public class TopicClusterPageStoreSizeTest extends JMSClusteredTestBase {
m2 = (TextMessage) cons1.receive(5000);
assertNotNull(m2, "got message");
assertTrue(m2.getJMSDestination().toString().contains(TOPIC));
- System.err.println("sub2 on 1, message txt:" + m2.getText());
+ logger.info("sub2 on 1, message txt:{}", m2.getText());
m2.acknowledge();
// if we send another, lb will kick in..
@@ -184,7 +195,7 @@ public class TopicClusterPageStoreSizeTest extends JMSClusteredTestBase {
assertNotNull(m2, "got message");
assertTrue(m2.getJMSDestination().toString().contains(TOPIC));
- System.err.println("sub2 on 2: message txt:" + m2.getText());
+ logger.info("sub2 on 2: message txt:{}", m2.getText());
m2.acknowledge();
// no duplicate, not available on cons1
@@ -207,7 +218,7 @@ public class TopicClusterPageStoreSizeTest extends JMSClusteredTestBase {
m2 = (TextMessage) cons2.receive(5000);
assertNotNull(m2, "got message");
assertTrue(m2.getJMSDestination().toString().contains(TOPIC));
- System.err.println("sub1 on: " + cf + " - message txt:" + m2.getText());
+ logger.info("sub1 on: {} - message txt: {}", cf, m2.getText());
m2.acknowledge();
conn2.close();
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/GlobalJournalPagingTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/GlobalJournalPagingTest.java
index bf5eb170f2..6c7d19c14c 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/GlobalJournalPagingTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/GlobalJournalPagingTest.java
@@ -129,7 +129,7 @@ public class GlobalJournalPagingTest extends JournalPagingTest {
}
Queue queue = server.locateQueue(ADDRESS);
- queue.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue.getPageSubscription().getPagingStore().forceAnotherPage(true);
sendFewMessages(numberOfMessages, session, producer, body);
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/JournalPagingTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/JournalPagingTest.java
index c1854e9000..21a2f05f6e 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/JournalPagingTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/JournalPagingTest.java
@@ -148,12 +148,12 @@ public class JournalPagingTest extends ActiveMQTestBase {
Queue queue = server.locateQueue(ADDRESS);
queue.getPagingStore().startPaging();
- queue.getPagingStore().forceAnotherPage(); // forcing an empty file, just to make it more challenging
+ queue.getPagingStore().forceAnotherPage(true); // forcing an empty file, just to make it more challenging
int page = 1;
for (int i = 0; i < numberOfMessages; i++) {
if (i % 10 == 0 && i > 0) {
- queue.getPagingStore().forceAnotherPage();
+ queue.getPagingStore().forceAnotherPage(true);
page++;
}
message = session.createMessage(true);
@@ -215,7 +215,7 @@ public class JournalPagingTest extends ActiveMQTestBase {
page4.delete(null);
page4.open(true);
for (int i = 0; i < 9; i++) {
- page4.write(messagesRead.get(i)); // this will make message 29 disappear
+ page4.write(messagesRead.get(i), true, false); // this will make message 29 disappear
}
page4.close(false);
@@ -455,7 +455,7 @@ public class JournalPagingTest extends ActiveMQTestBase {
if (i == 4) {
session.commit();
- queue.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue.getPageSubscription().getPagingStore().forceAnotherPage(true);
}
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/MessagesExpiredPagingTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/MessagesExpiredPagingTest.java
index 7951cdfa71..8f803cbc52 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/MessagesExpiredPagingTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/MessagesExpiredPagingTest.java
@@ -187,7 +187,7 @@ public class MessagesExpiredPagingTest extends ActiveMQTestBase {
producer.setTimeToLive(10);
c.consumedDelta.set(0);
}
- queues[0].getPagingStore().forceAnotherPage();
+ queues[0].getPagingStore().forceAnotherPage(true);
}
producer.send(session.createTextMessage("hello" + extraBody));
}
@@ -195,7 +195,7 @@ public class MessagesExpiredPagingTest extends ActiveMQTestBase {
producer.setTimeToLive(300);
for (int i = 0; i < numberOfMessageSecondWave; i++) {
if (i > 0 && i % pagingInterval == 0) {
- queues[0].getPagingStore().forceAnotherPage();
+ queues[0].getPagingStore().forceAnotherPage(true);
}
producer.send(session.createTextMessage("hello" + extraBody));
}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingReceiveTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingReceiveTest.java
index d4dc08ed78..81518714e9 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingReceiveTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingReceiveTest.java
@@ -114,7 +114,7 @@ public class PagingReceiveTest extends ActiveMQTestBase {
queue.getPageSubscription().getPagingStore().startPaging();
for (int i = 0; i < 10; i++) {
- queue.getPageSubscription().getPagingStore().forceAnotherPage();
+ queue.getPageSubscription().getPagingStore().forceAnotherPage(true);
}
locator.setBlockOnNonDurableSend(false).setBlockOnAcknowledge(false);
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/TestDeadlockOnPurgePagingTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/TestDeadlockOnPurgePagingTest.java
index 781fe950be..7d3010ba97 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/TestDeadlockOnPurgePagingTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/TestDeadlockOnPurgePagingTest.java
@@ -101,7 +101,7 @@ public class TestDeadlockOnPurgePagingTest extends ActiveMQTestBase {
}
if (i > 10 && i % 10 == 0) {
- purgeQueue.getPageSubscription().getPagingStore().forceAnotherPage();
+ purgeQueue.getPageSubscription().getPagingStore().forceAnotherPage(true);
}
}
session.commit();
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
index b14aa311d5..d97415f3dd 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/ReplicationTest.java
@@ -277,14 +277,14 @@ public final class ReplicationTest extends ActiveMQTestBase {
replicatedJournal.appendAddRecordTransactional(23, 24, (byte) 1, new FakeData());
PagedMessage pgmsg = new PagedMessageImpl(msg, new long[0]);
- manager.pageWrite(dummy, pgmsg, 1);
- manager.pageWrite(dummy, pgmsg, 2);
- manager.pageWrite(dummy, pgmsg, 3);
- manager.pageWrite(dummy, pgmsg, 4);
+ manager.pageWrite(dummy, pgmsg, 1, true);
+ manager.pageWrite(dummy, pgmsg, 2, true);
+ manager.pageWrite(dummy, pgmsg, 3, true);
+ manager.pageWrite(dummy, pgmsg, 4, true);
blockOnReplication(storage, manager);
- PagingManager pagingManager = createPageManager(backupServer.getStorageManager(), backupServer.getConfiguration(), backupServer.getExecutorFactory(), backupServer.getAddressSettingsRepository());
+ PagingManager pagingManager = createPageManager(backupServer.getStorageManager(), backupServer.getConfiguration(), backupServer.getExecutorFactory(), backupServer.getScheduledPool(), backupServer.getAddressSettingsRepository());
PagingStore store = pagingManager.getPageStore(dummy);
store.start();
@@ -717,9 +717,10 @@ public final class ReplicationTest extends ActiveMQTestBase {
protected PagingManager createPageManager(final StorageManager storageManager,
final Configuration configuration,
final ExecutorFactory executorFactory,
+ final ScheduledExecutorService scheduledExecutorService,
final HierarchicalRepository<AddressSettings> addressSettingsRepository) throws Exception {
- PagingManager paging = new PagingManagerImpl(new PagingStoreFactoryNIO(storageManager, configuration.getPagingLocation(), 1000, null, executorFactory, executorFactory, false, null), addressSettingsRepository, configuration.getManagementAddress());
+ PagingManager paging = new PagingManagerImpl(new PagingStoreFactoryNIO(storageManager, configuration.getPagingLocation(), 1000, scheduledExecutorService, executorFactory, false, null), addressSettingsRepository, configuration.getManagementAddress());
paging.start();
runAfter(paging::stop);
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java
index 751e4a8c93..d76915ae6f 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java
+++ b/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/replication/SharedNothingReplicationFlowControlTest.java
@@ -248,7 +248,7 @@ public class SharedNothingReplicationFlowControlTest extends ActiveMQTestBase {
if (i % 10 == 0) {
sess.commit();
- store.forceAnotherPage();
+ store.forceAnotherPage(true);
}
}
@@ -294,7 +294,7 @@ public class SharedNothingReplicationFlowControlTest extends ActiveMQTestBase {
ExecutorFactory executorFactory,
boolean syncNonTransactional,
IOCriticalErrorListener critialErrorListener) {
- super(storageManager, directory, syncTimeout, scheduledExecutor, executorFactory, executorFactory, syncNonTransactional, critialErrorListener);
+ super(storageManager, directory, syncTimeout, scheduledExecutor, executorFactory, syncNonTransactional, critialErrorListener);
}
PageStoreFactoryTestable(PagingStoreFactoryNIO other) {
diff --git a/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/FilteredPagingLeakTest.java b/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/FilteredPagingLeakTest.java
index 2ef9d791f4..537dfed964 100644
--- a/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/FilteredPagingLeakTest.java
+++ b/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/FilteredPagingLeakTest.java
@@ -132,7 +132,7 @@ public class FilteredPagingLeakTest extends AbstractLeakTest {
producer.send(message);
if (i > 0 && i % COMMIT_INTERVAL == 0) {
session.commit();
- pagingStore.forceAnotherPage();
+ pagingStore.forceAnotherPage(true);
}
}
session.commit();
diff --git a/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/PagingLeakTest.java b/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/PagingLeakTest.java
index d0e922c65d..e89cd7ac70 100644
--- a/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/PagingLeakTest.java
+++ b/tests/leak-tests/src/test/java/org/apache/activemq/artemis/tests/leak/PagingLeakTest.java
@@ -131,7 +131,7 @@ public class PagingLeakTest extends AbstractLeakTest {
producer.send(message);
if (i > 0 && i % COMMIT_INTERVAL == 0) {
session.commit();
- serverQueue.getPagingStore().forceAnotherPage();
+ serverQueue.getPagingStore().forceAnotherPage(true);
}
}
session.commit();
@@ -186,7 +186,7 @@ public class PagingLeakTest extends AbstractLeakTest {
producer.send(message);
if (i > 0 && i % COMMIT_INTERVAL == 0) {
session.commit();
- serverQueue.getPagingStore().forceAnotherPage();
+ serverQueue.getPagingStore().forceAnotherPage(true);
}
}
session.commit();
diff --git a/tests/performance-tests/src/test/java/org/apache/activemq/artemis/tests/performance/storage/PersistMultiThreadTest.java b/tests/performance-tests/src/test/java/org/apache/activemq/artemis/tests/performance/storage/PersistMultiThreadTest.java
index a43a34fd46..e5010424c5 100644
--- a/tests/performance-tests/src/test/java/org/apache/activemq/artemis/tests/performance/storage/PersistMultiThreadTest.java
+++ b/tests/performance-tests/src/test/java/org/apache/activemq/artemis/tests/performance/storage/PersistMultiThreadTest.java
@@ -286,6 +286,21 @@ public class PersistMultiThreadTest extends ActiveMQTestBase {
}
+ @Override
+ public void readLock() {
+
+ }
+
+ @Override
+ public boolean readLock(long timeout) {
+ return true;
+ }
+
+ @Override
+ public void readUnlock() {
+
+ }
+
@Override
public void counterSnapshot() {
}
@@ -400,11 +415,6 @@ public class PersistMultiThreadTest extends ActiveMQTestBase {
return false;
}
- @Override
- public void addSyncPoint(OperationContext context) throws Exception {
-
- }
-
@Override
public Page usePage(long page, boolean create) {
return null;
@@ -486,7 +496,7 @@ public class PersistMultiThreadTest extends ActiveMQTestBase {
}
@Override
- public void forceAnotherPage() throws Exception {
+ public void forceAnotherPage(boolean useExecutor) throws Exception {
}
@@ -502,7 +512,6 @@ public class PersistMultiThreadTest extends ActiveMQTestBase {
@Override
public void stopPaging() throws Exception {
-
}
@Override
@@ -525,12 +534,16 @@ public class PersistMultiThreadTest extends ActiveMQTestBase {
}
@Override
- public boolean lock(long timeout) {
+ public void writeLock() {
+ }
+
+ @Override
+ public boolean writeLock(long timeout) {
return false;
}
@Override
- public void unlock() {
+ public void writeUnlock() {
}
diff --git a/tests/soak-tests/src/main/resources/servers/multipleProducersPagingSoak/broker.xml b/tests/soak-tests/src/main/resources/servers/multipleProducersPagingSoak/broker.xml
new file mode 100644
index 0000000000..699bd30a57
--- /dev/null
+++ b/tests/soak-tests/src/main/resources/servers/multipleProducersPagingSoak/broker.xml
@@ -0,0 +1,191 @@
+<?xml version='1.0'?>
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+<configuration xmlns="urn:activemq"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="urn:activemq /schema/artemis-configuration.xsd">
+
+ <core xmlns="urn:activemq:core" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="urn:activemq:core ">
+
+ <name>0.0.0.0</name>
+
+ <persistence-enabled>true</persistence-enabled>
+
+ <!-- this could be ASYNCIO or NIO
+ -->
+ <journal-type>NIO</journal-type>
+
+ <paging-directory>./data/paging</paging-directory>
+
+ <bindings-directory>./data/bindings</bindings-directory>
+
+ <journal-directory>./data/journal</journal-directory>
+
+ <large-messages-directory>./data/large-messages</large-messages-directory>
+
+ <journal-datasync>true</journal-datasync>
+
+ <journal-min-files>2</journal-min-files>
+
+ <journal-pool-files>-1</journal-pool-files>
+
+ <message-expiry-scan-period>1000</message-expiry-scan-period>
+
+ <!--
+ You can verify the network health of a particular NIC by specifying the <network-check-NIC> element.
+ <network-check-NIC>theNicName</network-check-NIC>
+ -->
+
+ <!--
+ Use this to use an HTTP server to validate the network
+ <network-check-URL-list>http://www.apache.org</network-check-URL-list> -->
+
+ <!-- <network-check-period>10000</network-check-period> -->
+ <!-- <network-check-timeout>1000</network-check-timeout> -->
+
+ <!-- this is a comma separated list, no spaces, just DNS or IPs
+ it should accept IPV6
+
+ Warning: Make sure you understand your network topology as this is meant to validate if your network is valid.
+ Using IPs that could eventually disappear or be partially visible may defeat the purpose.
+ You can use a list of multiple IPs, and if any successful ping will make the server OK to continue running -->
+ <!-- <network-check-list>10.0.0.1</network-check-list> -->
+
+ <!-- use this to customize the ping used for ipv4 addresses -->
+ <!-- <network-check-ping-command>ping -c 1 -t %d %s</network-check-ping-command> -->
+
+ <!-- use this to customize the ping used for ipv6 addresses -->
+ <!-- <network-check-ping6-command>ping6 -c 1 %2$s</network-check-ping6-command> -->
+
+
+
+
+ <!-- how often we are looking for how many bytes are being used on the disk in ms -->
+ <disk-scan-period>5000</disk-scan-period>
+
+ <page-sync-timeout>100</page-sync-timeout>
+
+
+ <!-- once the disk hits this limit the system will block, or close the connection in certain protocols
+ that won't support flow control. -->
+ <max-disk-usage>90</max-disk-usage>
+
+ <!-- the system will enter into page mode once you hit this limit.
+ This is an estimate in bytes of how much the messages are using in memory
+
+ The system will use half of the available memory (-Xmx) by default for the global-max-size.
+ You may specify a different value here if you need to customize it to your needs.
+
+ <global-max-size>100Mb</global-max-size>
+
+ -->
+
+ <acceptors>
+
+ <!-- useEpoll means: it will use Netty epoll if you are on a system (Linux) that supports it -->
+ <!-- useKQueue means: it will use Netty kqueue if you are on a system (MacOS) that supports it -->
+ <!-- amqpCredits: The number of credits sent to AMQP producers -->
+ <!-- amqpLowCredits: The server will send the # credits specified at amqpCredits at this low mark -->
+
+ <!-- Acceptor for every supported protocol -->
+ <acceptor name="artemis">tcp://0.0.0.0:61616?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=CORE,AMQP,STOMP,HORNETQ,MQTT,OPENWIRE;useEpoll=true;useKQueue;amqpCredits=1000;amqpLowCredits=300;actorThresholdBytes=10000</acceptor>
+
+ <!-- AMQP Acceptor. Listens on default AMQP port for AMQP traffic.-->
+ <acceptor name="amqp">tcp://0.0.0.0:5672?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=AMQP;useEpoll=true;useKQueue=true;amqpCredits=1000;amqpLowCredits=300</acceptor>
+
+ <!-- STOMP Acceptor. -->
+ <acceptor name="stomp">tcp://0.0.0.0:61613?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=STOMP;useEpoll=true;useKQueue=true</acceptor>
+
+ <!-- HornetQ Compatibility Acceptor. Enables HornetQ Core and STOMP for legacy HornetQ clients. -->
+ <acceptor name="hornetq">tcp://0.0.0.0:5445?protocols=HORNETQ,STOMP;useEpoll=true;useKQueue=true</acceptor>
+
+ <!-- MQTT Acceptor -->
+ <acceptor name="mqtt">tcp://0.0.0.0:1883?tcpSendBufferSize=1048576;tcpReceiveBufferSize=1048576;protocols=MQTT;useEpoll=true;useKQueue=true</acceptor>
+
+ </acceptors>
+
+
+ <security-settings>
+ <security-setting match="#">
+ <permission type="createNonDurableQueue" roles="guest"/>
+ <permission type="deleteNonDurableQueue" roles="guest"/>
+ <permission type="createDurableQueue" roles="guest"/>
+ <permission type="deleteDurableQueue" roles="guest"/>
+ <permission type="createAddress" roles="guest"/>
+ <permission type="deleteAddress" roles="guest"/>
+ <permission type="consume" roles="guest"/>
+ <permission type="browse" roles="guest"/>
+ <permission type="send" roles="guest"/>
+ <!-- we need this otherwise ./artemis data imp wouldn't work -->
+ <permission type="manage" roles="guest"/>
+ </security-setting>
+ </security-settings>
+
+ <address-settings>
+ <!-- if you define auto-create on certain queues, management has to be auto-create -->
+ <address-setting match="activemq.management#">
+ <dead-letter-address>DLQ</dead-letter-address>
+ <expiry-address>ExpiryQueue</expiry-address>
+ <redelivery-delay>0</redelivery-delay>
+ <!-- with -1 only the global-max-size is in use for limiting -->
+ <max-size-bytes>-1</max-size-bytes>
+ <message-counter-history-day-limit>10</message-counter-history-day-limit>
+ <address-full-policy>PAGE</address-full-policy>
+ <auto-create-queues>true</auto-create-queues>
+ <auto-create-addresses>true</auto-create-addresses>
+ </address-setting>
+ <!--default for catch all-->
+ <address-setting match="#">
+ <dead-letter-address>DLQ</dead-letter-address>
+ <expiry-address>ExpiryQueue</expiry-address>
+ <redelivery-delay>0</redelivery-delay>
+ <max-read-page-messages>-1</max-read-page-messages>
+ <max-read-page-bytes>-1</max-read-page-bytes>
+ <max-size-bytes>0</max-size-bytes>
+ <page-size-bytes>10M</page-size-bytes>
+ <message-counter-history-day-limit>10</message-counter-history-day-limit>
+ <address-full-policy>PAGE</address-full-policy>
+ <auto-create-queues>true</auto-create-queues>
+ <auto-create-addresses>true</auto-create-addresses>
+ </address-setting>
+ </address-settings>
+
+ <addresses>
+ <address name="DLQ">
+ <anycast>
+ <queue name="DLQ" />
+ </anycast>
+ </address>
+ <address name="ExpiryQueue">
+ <anycast>
+ <queue name="ExpiryQueue" />
+ </anycast>
+ </address>
+ <address name="MultipleProducers">
+ <anycast>
+ <queue name="MultipleProducers" />
+ </anycast>
+ </address>
+
+ </addresses>
+
+ </core>
+</configuration>
diff --git a/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/paging/MultipleProducersPagingSoakTest.java b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/paging/MultipleProducersPagingSoakTest.java
new file mode 100644
index 0000000000..f076809de2
--- /dev/null
+++ b/tests/soak-tests/src/test/java/org/apache/activemq/artemis/tests/soak/paging/MultipleProducersPagingSoakTest.java
@@ -0,0 +1,190 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.artemis.tests.soak.paging;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Queue;
+import javax.jms.Session;
+import java.io.File;
+import java.lang.invoke.MethodHandles;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.activemq.artemis.cli.commands.helper.HelperCreate;
+import org.apache.activemq.artemis.tests.soak.SoakTestBase;
+import org.apache.activemq.artemis.tests.util.CFUtil;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import static org.apache.activemq.artemis.utils.TestParameters.testProperty;
+
+public class MultipleProducersPagingSoakTest extends SoakTestBase {
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ public static final String SERVER_NAME_0 = "multiple-producers-paging";
+
+ public static final String TEST_NAME = "MULTIPLE_PRODUCERS";
+
+ @BeforeAll
+ public static void createServers() throws Exception {
+
+ File server0Location = getFileServerLocation(SERVER_NAME_0);
+ deleteDirectory(server0Location);
+
+ {
+ HelperCreate cliCreateServer = helperCreate();
+ cliCreateServer.setUser("admin").setPassword("admin").setAllowAnonymous(true).setNoWeb(false).setArtemisInstance(server0Location).setConfiguration("./src/main/resources/servers/multipleProducersPagingSoak");
+ cliCreateServer.createServer();
+ }
+ }
+
+ @BeforeEach
+ public void before() throws Exception {
+ cleanupData(SERVER_NAME_0);
+ startServer(SERVER_NAME_0, 0, 30000);
+ }
+
+
+ @Test
+ public void testMultipleProducers() throws Exception {
+ String protocol = "amqp";
+ int producers = testProperty(TEST_NAME, "PRODUCERS", 100);
+ int consumers = testProperty(TEST_NAME, "CONSUMERS", 10);
+ int messagesPerProducer = testProperty(TEST_NAME, "MESSAGES_PER_PRODUCER", 250);
+ int totalMessages = producers * messagesPerProducer;
+ assertTrue(totalMessages % consumers == 0, "totalMessages % consumers must be 0");
+ int messagesPerConsumer = totalMessages / consumers;
+
+ String queueName = "MultipleProducers";
+ AtomicInteger errors = new AtomicInteger();
+
+ CountDownLatch done = new CountDownLatch(producers + consumers);
+ AtomicInteger distance = new AtomicInteger(0);
+
+ final AtomicInteger[] messagesSent = new AtomicInteger[producers];
+ for (int i = 0; i < messagesSent.length; i++) {
+ messagesSent[i] = new AtomicInteger(0);
+ }
+
+ final AtomicInteger[] messagesConsumed = new AtomicInteger[consumers];
+ for (int i = 0; i < messagesConsumed.length; i++) {
+ messagesConsumed[i] = new AtomicInteger(0);
+ }
+
+ ExecutorService executor = Executors.newFixedThreadPool(producers + consumers + 1);
+ runAfter(executor::shutdownNow);
+
+ AtomicBoolean running = new AtomicBoolean(true);
+
+ executor.execute(() -> {
+ try {
+ while (running.get()) {
+ done.await(1, TimeUnit.SECONDS);
+ StringBuilder builder = new StringBuilder();
+ int produced = 0, consumed = 0;
+ for (int i = 0; i < producers; i++) {
+ builder.append("Producer[" + i + "] sent " + messagesSent[i] + "\n");
+ produced += messagesSent[i].get();
+ }
+ for (int i = 0; i < consumers; i++) {
+ builder.append("Consumer[" + i + "] received " + messagesConsumed[i] + "\n");
+ consumed += messagesConsumed[i].get();
+ }
+ builder.append("Total produced: " + produced + "\n");
+ builder.append("Total consumed: " + consumed + "\n");
+ logger.info("\n{}", builder.toString());
+ }
+ } catch (InterruptedException expected) {
+ }
+ });
+
+ for (int i = 0; i < producers; i++) {
+ int producerID = i;
+
+
+ ConnectionFactory factory = CFUtil.createConnectionFactory(protocol, "tcp://localhost:61616");
+ executor.execute(() -> {
+ try (Connection connection = factory.createConnection()) {
+ try (Session session = connection.createSession(true, Session.SESSION_TRANSACTED)) {
+ Queue queue = session.createQueue(queueName);
+ MessageProducer producer = session.createProducer(queue);
+ for (int produced = 0; produced < messagesPerProducer; produced++) {
+ producer.send(session.createTextMessage("hello hello"));
+ session.commit();
+ messagesSent[producerID].incrementAndGet();
+ distance.incrementAndGet();
+ }
+ done.countDown();
+ }
+ } catch (Exception e) {
+ logger.warn(e.getMessage(), e);
+ errors.incrementAndGet();
+ }
+ });
+ }
+ for (int i = 0; i < consumers; i++) {
+ int consumerID = i;
+
+ executor.execute(() -> {
+ ConnectionFactory factory = CFUtil.createConnectionFactory(protocol, "tcp://localhost:61616");
+ try (Connection connection = factory.createConnection()) {
+ connection.start();
+ Session session = connection.createSession(true, Session.SESSION_TRANSACTED);
+ Queue queue = session.createQueue(queueName);
+ MessageConsumer consumer = session.createConsumer(queue);
+ for (int consumed = 0; consumed < messagesPerConsumer; consumed++) {
+ Message message = consumer.receive(10_000);
+ assertNotNull(message);
+ messagesConsumed[consumerID].incrementAndGet();
+ distance.decrementAndGet();
+ if (consumed % 100 == 0) {
+ session.commit();
+ }
+ }
+ session.commit();
+ } catch (Throwable e) {
+ errors.incrementAndGet();
+ logger.warn(e.getMessage(), e);
+ } finally {
+ done.countDown();
+ }
+ });
+ }
+
+ assertTrue(done.await(10, TimeUnit.MINUTES));
+ assertEquals(0, errors.get());
+
+ }
+
+
+}
diff --git a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/lockmanager/LockManagerPageCleanupWhileReplicaCatchupTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManagerAccessor.java
similarity index 53%
rename from tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/lockmanager/LockManagerPageCleanupWhileReplicaCatchupTest.java
rename to tests/unit-tests/src/test/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManagerAccessor.java
index f295a53dca..1bd4f90d0c 100644
--- a/tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/cluster/failover/lockmanager/LockManagerPageCleanupWhileReplicaCatchupTest.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/core/persistence/impl/journal/JournalStorageManagerAccessor.java
@@ -14,21 +14,15 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.activemq.artemis.tests.integration.cluster.failover.lockmanager;
-import org.apache.activemq.artemis.core.config.ha.ReplicationBackupPolicyConfiguration;
-import org.apache.activemq.artemis.tests.integration.cluster.failover.PageCleanupWhileReplicaCatchupTest;
+package org.apache.activemq.artemis.core.persistence.impl.journal;
-public class LockManagerPageCleanupWhileReplicaCatchupTest extends PageCleanupWhileReplicaCatchupTest {
+import org.apache.activemq.artemis.core.replication.ReplicationManager;
- @Override
- protected void createConfigs() throws Exception {
- createPluggableReplicatedConfigs();
- }
+public class JournalStorageManagerAccessor {
- @Override
- protected void setupHAPolicyConfiguration() {
- ((ReplicationBackupPolicyConfiguration) backupConfig.getHAPolicyConfiguration()).setMaxSavedReplicatedJournalsSize(2).setAllowFailBack(true);
+ public static void setReplicationManager(JournalStorageManager journalStorageManager, ReplicationManager replicationManager) {
+ journalStorageManager.setReplicator(replicationManager);
}
}
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/AmqpPageTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/AmqpPageTest.java
index 4196a9d49a..6949305d89 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/AmqpPageTest.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/AmqpPageTest.java
@@ -69,10 +69,10 @@ public class AmqpPageTest extends PageTest {
Page page) throws Exception {
if (!isLargeMessage) {
final Message message = createStandardMessage(address, msgID, content);
- page.write(new PagedMessageImpl(message, new long[0]));
+ page.write(new PagedMessageImpl(message, new long[0]), true, false);
} else {
final AMQPLargeMessage message = createLargeMessage(storageManager, address, msgID, content);
- page.write(new PagedMessageImpl(message, new long[0]));
+ page.write(new PagedMessageImpl(message, new long[0]), true, false);
message.releaseResources(false, false);
}
}
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PageTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PageTest.java
index bf905744db..30e3fe667f 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PageTest.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PageTest.java
@@ -364,14 +364,14 @@ public class PageTest extends ActiveMQTestBase {
msg.setMessageID(msgID);
msg.addBytes(content);
msg.setAddress(address);
- page.write(new PagedMessageImpl(msg, new long[0]));
+ page.write(new PagedMessageImpl(msg, new long[0]), true, false);
msg.releaseResources(false, false);
} else {
ICoreMessage msg = new CoreMessage().initBuffer(100);
msg.setMessageID(msgID);
msg.getBodyBuffer().writeBytes(content);
msg.setAddress(address);
- page.write(new PagedMessageImpl(msg, new long[0]));
+ page.write(new PagedMessageImpl(msg, new long[0]), true, false);
}
}
}
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PageTimedWriterUnitTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PageTimedWriterUnitTest.java
new file mode 100644
index 0000000000..47d40e4889
--- /dev/null
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PageTimedWriterUnitTest.java
@@ -0,0 +1,532 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.activemq.artemis.tests.unit.core.paging.impl;
+
+import java.lang.invoke.MethodHandles;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.activemq.artemis.api.core.ActiveMQBuffer;
+import org.apache.activemq.artemis.api.core.SimpleString;
+import org.apache.activemq.artemis.core.config.Configuration;
+import org.apache.activemq.artemis.core.config.impl.ConfigurationImpl;
+import org.apache.activemq.artemis.core.io.IOCallback;
+import org.apache.activemq.artemis.core.io.OperationConsistencyLevel;
+import org.apache.activemq.artemis.core.io.SequentialFileFactory;
+import org.apache.activemq.artemis.core.journal.Journal;
+import org.apache.activemq.artemis.core.message.impl.CoreMessage;
+import org.apache.activemq.artemis.core.paging.PagedMessage;
+import org.apache.activemq.artemis.core.paging.PagingManager;
+import org.apache.activemq.artemis.core.paging.PagingStoreFactory;
+import org.apache.activemq.artemis.core.paging.impl.PageTimedWriter;
+import org.apache.activemq.artemis.core.paging.impl.PagedMessageImpl;
+import org.apache.activemq.artemis.core.paging.impl.PagingStoreImpl;
+import org.apache.activemq.artemis.core.persistence.OperationContext;
+import org.apache.activemq.artemis.core.persistence.impl.journal.JournalStorageManager;
+import org.apache.activemq.artemis.core.persistence.impl.journal.JournalStorageManagerAccessor;
+import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
+import org.apache.activemq.artemis.core.replication.ReplicationManager;
+import org.apache.activemq.artemis.core.server.JournalType;
+import org.apache.activemq.artemis.core.server.Queue;
+import org.apache.activemq.artemis.core.server.RouteContextList;
+import org.apache.activemq.artemis.core.server.impl.RoutingContextImpl;
+import org.apache.activemq.artemis.core.settings.impl.AddressFullMessagePolicy;
+import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
+import org.apache.activemq.artemis.core.transaction.Transaction;
+import org.apache.activemq.artemis.core.transaction.TransactionOperationAbstract;
+import org.apache.activemq.artemis.core.transaction.impl.TransactionImpl;
+import org.apache.activemq.artemis.tests.unit.core.journal.impl.fakes.FakeSequentialFileFactory;
+import org.apache.activemq.artemis.tests.util.ArtemisTestCase;
+import org.apache.activemq.artemis.utils.ExecutorFactory;
+import org.apache.activemq.artemis.utils.RandomUtil;
+import org.apache.activemq.artemis.utils.ReusableLatch;
+import org.apache.activemq.artemis.utils.actors.OrderedExecutorFactory;
+import org.apache.activemq.artemis.utils.critical.CriticalAnalyzer;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class PageTimedWriterUnitTest extends ArtemisTestCase {
+
+ private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
+
+ private static final SimpleString ADDRESS = SimpleString.of("someAddress");
+
+ ScheduledExecutorService scheduledExecutorService;
+ ExecutorService executorService;
+ OrderedExecutorFactory executorFactory;
+ OperationContext context;
+
+ RouteContextList routeContextList;
+
+ // almost real as we use an extension that can allow mocking internal objects such as journals and replicationManager
+ JournalStorageManager realJournalStorageManager;
+
+ PagingStoreImpl pageStore;
+
+ CountDownLatch allowRunning;
+
+ final ReusableLatch enteredSync = new ReusableLatch(1);
+ final ReusableLatch allowSync = new ReusableLatch(0);
+ final AtomicInteger doneSync = new AtomicInteger(0);
+
+ final AtomicInteger pageWrites = new AtomicInteger(0);
+
+ PageTimedWriter timer;
+
+ Configuration configuration;
+
+ SequentialFileFactory inMemoryFileFactory;
+
+ Journal mockBindingsJournal;
+ Journal mockMessageJournal;
+
+ AtomicBoolean useReplication = new AtomicBoolean(false);
+ AtomicBoolean returnSynchronizing = new AtomicBoolean(false);
+ ReplicationManager mockReplicationManager;
+
+
+ private class MockableJournalStorageManager extends JournalStorageManager {
+
+ MockableJournalStorageManager(Configuration config,
+ Journal bindingsJournal,
+ Journal messagesJournal,
+ ExecutorFactory executorFactory,
+ ExecutorFactory ioExecutors) {
+ super(config, Mockito.mock(CriticalAnalyzer.class), executorFactory, ioExecutors);
+ this.bindingsJournal = bindingsJournal;
+ this.messageJournal = messagesJournal;
+ }
+
+ @Override
+ public void start() throws Exception {
+ super.start();
+ idGenerator.forceNextID(1);
+ }
+
+ @Override
+ protected void createDirectories() {
+ // not creating any folders
+ }
+
+ @Override
+ public void pageWrite(final SimpleString address, final PagedMessage message, final long pageNumber, boolean storageUp, boolean originallyReplicated) {
+ super.pageWrite(address, message, pageNumber, storageUp, originallyReplicated);
+ pageWrites.incrementAndGet();
+ }
+ }
+
+ private CoreMessage createMessage() {
+ long id = realJournalStorageManager.generateID();
+ ActiveMQBuffer buffer = RandomUtil.randomBuffer(10);
+ CoreMessage msg = new CoreMessage(id, 50 + buffer.capacity());
+
+ msg.setAddress(ADDRESS);
+
+ msg.getBodyBuffer().resetReaderIndex();
+ msg.getBodyBuffer().resetWriterIndex();
+
+ msg.getBodyBuffer().writeBytes(buffer, buffer.capacity());
+
+ return msg;
+ }
+
+ @BeforeEach
+ public void setupMocks() throws Exception {
+ configuration = new ConfigurationImpl();
+ configuration.setJournalType(JournalType.NIO);
+ scheduledExecutorService = Executors.newScheduledThreadPool(10);
+ executorService = Executors.newFixedThreadPool(10);
+ runAfter(scheduledExecutorService::shutdownNow);
+ runAfter(executorService::shutdownNow);
+ runAfter(() -> OperationContextImpl.clearContext());
+ executorFactory = new OrderedExecutorFactory(executorService);
+ context = OperationContextImpl.getContext(executorFactory);
+ assertNotNull(context);
+
+ mockBindingsJournal = Mockito.mock(Journal.class);
+ mockMessageJournal = Mockito.mock(Journal.class);
+
+ mockReplicationManager = Mockito.mock(ReplicationManager.class);
+ Mockito.when(mockReplicationManager.isStarted()).thenAnswer(a -> useReplication.get());
+ Mockito.when(mockReplicationManager.isSynchronizing()).thenAnswer(a -> returnSynchronizing.get());
+ Mockito.doAnswer(a -> {
+ if (useReplication.get()) {
+ OperationContext ctx = OperationContextImpl.getContext();
+ if (ctx != null) {
+ ctx.replicationDone();
+ }
+ }
+ return null;
+ }).when(mockReplicationManager).pageWrite(Mockito.any(SimpleString.class), Mockito.any(PagedMessage.class), Mockito.anyLong(), Mockito.anyBoolean());
+
+ realJournalStorageManager = new MockableJournalStorageManager(configuration, mockBindingsJournal, mockMessageJournal, executorFactory, executorFactory);
+ realJournalStorageManager.start();
+
+ JournalStorageManagerAccessor.setReplicationManager(realJournalStorageManager, mockReplicationManager);
+
+ allowRunning = new CountDownLatch(1);
+
+ inMemoryFileFactory = new FakeSequentialFileFactory();
+
+ PagingStoreFactory mockPageStoreFactory = Mockito.mock(PagingStoreFactory.class);
+ PagingManager mockPagingManager = Mockito.mock(PagingManager.class);
+
+ final int MAX_SIZE = 1024 * 10;
+
+ AddressSettings settings = new AddressSettings().setPageSizeBytes(MAX_SIZE).setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
+
+ pageStore = new PagingStoreImpl(ADDRESS, scheduledExecutorService, 100, mockPagingManager, realJournalStorageManager, inMemoryFileFactory,
+ mockPageStoreFactory, ADDRESS, settings, executorFactory.getExecutor(), false) {
+ @Override
+ protected PageTimedWriter createPageTimedWriter(ScheduledExecutorService scheduledExecutor, long syncTimeout) {
+
+ PageTimedWriter timer = new PageTimedWriter(100_000, realJournalStorageManager, this, scheduledExecutorService, executorFactory.getExecutor(), true, 100) {
+ @Override
+ public void run() {
+ try {
+ allowRunning.await();
+ } catch (InterruptedException e) {
+ logger.warn(e.getMessage(), e);
+ Thread.currentThread().interrupt();
+
+ }
+ super.run();
+ }
+
+ @Override
+ protected void performSync() throws Exception {
+ enteredSync.countDown();
+ super.performSync();
+ try {
+ allowSync.await();
+ } catch (InterruptedException e) {
+ logger.warn(e.getMessage(), e);
+ }
+ doneSync.incrementAndGet();
+ }
+ };
+
+ timer.start();
+ return timer;
+
+ }
+ };
+
+ timer = pageStore.getPageTimedWriter();
+
+ pageStore.start();
+ pageStore.startPaging();
+
+ routeContextList = new RoutingContextImpl.ContextListing();
+ Queue mockQueue = Mockito.mock(Queue.class);
+ Mockito.when(mockQueue.getID()).thenReturn(1L);
+ routeContextList.addAckedQueue(mockQueue);
+ }
+
+ // a test to validate if the Mocks are correctly setup
+ @Test
+ public void testValidateMocks() throws Exception {
+ TransactionImpl tx = new TransactionImpl(realJournalStorageManager);
+ tx.setContainsPersistent();
+ AtomicInteger count = new AtomicInteger(0);
+ tx.addOperation(new TransactionOperationAbstract() {
+ @Override
+ public void afterCommit(Transaction tx) {
+ count.incrementAndGet();
+ }
+ });
+ assertEquals(0, count.get());
+ tx.commit();
+ assertEquals(1, count.get(), "tx.commit is not correctly wired on mocking");
+
+
+ realJournalStorageManager.afterCompleteOperations(new IOCallback() {
+ @Override
+ public void done() {
+ count.incrementAndGet();
+ }
+
+ @Override
+ public void onError(int errorCode, String errorMessage) {
+
+ }
+ });
+
+ realJournalStorageManager.afterCompleteOperations(new IOCallback() {
+ @Override
+ public void done() {
+ count.incrementAndGet();
+ }
+
+ @Override
+ public void onError(int errorCode, String errorMessage) {
+
+ }
+ }, OperationConsistencyLevel.FULL);
+
+ assertEquals(3, count.get(), "afterCompletion is not correctly wired on mocking");
+
+ long id = realJournalStorageManager.generateID();
+ long newID = realJournalStorageManager.generateID();
+ assertEquals(1L, newID - id);
+
+ }
+
+ PagedMessage createPagedMessage() {
+ return new PagedMessageImpl(createMessage(), new long[] {1}, -1);
+ }
+
+ @Test
+ public void testIOCompletion() throws Exception {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ timer.addTask(context, createPagedMessage(), null, Mockito.mock(RouteContextList.class));
+
+ context.executeOnCompletion(new IOCallback() {
+ @Override
+ public void done() {
+ latch.countDown();
+ }
+
+ @Override
+ public void onError(int errorCode, String errorMessage) {
+ }
+ });
+
+ assertFalse(latch.await(10, TimeUnit.MILLISECONDS));
+ allowRunning.countDown();
+ assertTrue(latch.await(1, TimeUnit.MINUTES));
+
+ }
+
+ @Test
+ public void testIOCompletionWhileReplica() throws Exception {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ useReplication.set(true);
+
+ timer.addTask(context, createPagedMessage(), null, Mockito.mock(RouteContextList.class));
+
+ context.executeOnCompletion(new IOCallback() {
+ @Override
+ public void done() {
+ latch.countDown();
+ }
+
+ @Override
+ public void onError(int errorCode, String errorMessage) {
+ }
+ });
+
+ assertFalse(latch.await(10, TimeUnit.MILLISECONDS));
+ allowRunning.countDown();
+ assertTrue(latch.await(10, TimeUnit.MINUTES));
+ }
+
+ @Test
+ public void testTXCompletionWhileReplica() throws Exception {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ useReplication.set(true);
+
+ TransactionImpl transaction = new TransactionImpl(realJournalStorageManager);
+ transaction.setContainsPersistent();
+ transaction.addOperation(new TransactionOperationAbstract() {
+ @Override
+ public void afterCommit(Transaction tx) {
+ super.afterCommit(tx);
+ latch.countDown();
+ }
+ });
+
+ timer.addTask(context, createPagedMessage(), transaction, Mockito.mock(RouteContextList.class));
+
+ transaction.commit();
+
+ assertFalse(latch.await(10, TimeUnit.MILLISECONDS));
+ allowRunning.countDown();
+ assertTrue(latch.await(10, TimeUnit.MINUTES));
+ }
+
+ @Test
+ public void testAllowWriteWhileSyncPending() throws Exception {
+ int numberOfMessages = 100;
+ CountDownLatch latch = new CountDownLatch(numberOfMessages);
+
+ allowRunning.countDown();
+ useReplication.set(false);
+
+ assertEquals(0, doneSync.get());
+ enteredSync.setCount(1);
+ allowSync.setCount(1);
+
+ for (int i = 0; i < numberOfMessages; i++) {
+ TransactionImpl newTX = new TransactionImpl(realJournalStorageManager);
+ newTX.setContainsPersistent();
+ newTX.addOperation(new TransactionOperationAbstract() {
+ @Override
+ public void afterCommit(Transaction tx) {
+ super.afterCommit(tx);
+ latch.countDown();
+ }
+ });
+ pageStore.page(createMessage(), newTX, routeContextList);
+ newTX.commit();
+
+ if (i == 0) {
+ assertTrue(enteredSync.await(10, TimeUnit.SECONDS));
+ assertEquals(0, doneSync.get()); // it should not complete until allowSync is released
+ }
+ }
+
+ assertFalse(latch.await(10, TimeUnit.MILLISECONDS));
+ allowSync.countDown();
+ assertTrue(latch.await(10, TimeUnit.MINUTES));
+
+ assertEquals(numberOfMessages, pageWrites.get());
+ }
+
+ @Test
+ public void testTXCompletionWhileDisableReplica() throws Exception {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ useReplication.set(true);
+
+ TransactionImpl transaction = new TransactionImpl(realJournalStorageManager);
+ transaction.setContainsPersistent();
+ transaction.addOperation(new TransactionOperationAbstract() {
+ @Override
+ public void afterCommit(Transaction tx) {
+ super.afterCommit(tx);
+ latch.countDown();
+ }
+ });
+
+ timer.addTask(context, createPagedMessage(), transaction, Mockito.mock(RouteContextList.class));
+
+ transaction.commit();
+
+ assertFalse(latch.await(10, TimeUnit.MILLISECONDS));
+ useReplication.set(false);
+ allowRunning.countDown();
+ assertTrue(latch.await(10, TimeUnit.MINUTES));
+ }
+
+ // add a task while replicating, process it when no longer replicating (disconnect a node scenario)
+ @Test
+ public void testDisableReplica() throws Exception {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ useReplication.set(true);
+
+ timer.addTask(context, createPagedMessage(), null, Mockito.mock(RouteContextList.class));
+
+ context.executeOnCompletion(new IOCallback() {
+ @Override
+ public void done() {
+ latch.countDown();
+ }
+
+ @Override
+ public void onError(int errorCode, String errorMessage) {
+ }
+ });
+
+ assertFalse(latch.await(10, TimeUnit.MILLISECONDS));
+ useReplication.set(false);
+ allowRunning.countDown();
+ assertTrue(latch.await(10, TimeUnit.SECONDS));
+ }
+
+ // add a task while not replicating, process it when is now replicating (reconnect a node scenario)
+ // this is the oppostie from testDisableReplica
+ @Test
+ public void testEnableReplica() throws Exception {
+ CountDownLatch latch = new CountDownLatch(1);
+
+ useReplication.set(false);
+
+ timer.addTask(context, createPagedMessage(), null, Mockito.mock(RouteContextList.class));
+
+ context.executeOnCompletion(new IOCallback() {
+ @Override
+ public void done() {
+ latch.countDown();
+ }
+
+ @Override
+ public void onError(int errorCode, String errorMessage) {
+ }
+ });
+
+ assertFalse(latch.await(10, TimeUnit.MILLISECONDS));
+ useReplication.set(true);
+ allowRunning.countDown();
+ assertTrue(latch.await(10, TimeUnit.SECONDS));
+ }
+
+ @Test
+ public void testTXCompletion() throws Exception {
+ ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(10);
+ ExecutorService executorService = Executors.newFixedThreadPool(10);
+ runAfter(scheduledExecutorService::shutdownNow);
+ runAfter(executorService::shutdownNow);
+ runAfter(() -> OperationContextImpl.clearContext());
+
+ OrderedExecutorFactory executorFactory = new OrderedExecutorFactory(executorService);
+
+ OperationContextImpl.clearContext();
+
+ OperationContext context = OperationContextImpl.getContext(executorFactory);
+
+ CountDownLatch latch = new CountDownLatch(1);
+
+ Transaction tx = new TransactionImpl(realJournalStorageManager, Integer.MAX_VALUE);
+ tx.setContainsPersistent();
+
+ timer.addTask(context, createPagedMessage(), tx, Mockito.mock(RouteContextList.class));
+ tx.addOperation(new TransactionOperationAbstract() {
+ @Override
+ public void afterCommit(Transaction tx) {
+ latch.countDown();
+ }
+ });
+ tx.commit();
+
+ assertFalse(latch.await(10, TimeUnit.MILLISECONDS));
+
+ allowRunning.countDown();
+
+ assertTrue(latch.await(10, TimeUnit.SECONDS));
+
+ }
+
+}
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PagingManagerImplTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PagingManagerImplTest.java
index 5954e14c28..b7c72fa03a 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PagingManagerImplTest.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PagingManagerImplTest.java
@@ -18,6 +18,9 @@ package org.apache.activemq.artemis.tests.unit.core.paging.impl;
import java.io.File;
import java.nio.ByteBuffer;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
import org.apache.activemq.artemis.api.core.ICoreMessage;
import org.apache.activemq.artemis.api.core.SimpleString;
@@ -28,6 +31,7 @@ import org.apache.activemq.artemis.core.paging.impl.Page;
import org.apache.activemq.artemis.core.paging.impl.PagingManagerImpl;
import org.apache.activemq.artemis.core.paging.impl.PagingStoreFactoryNIO;
import org.apache.activemq.artemis.core.persistence.StorageManager;
+import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
import org.apache.activemq.artemis.core.persistence.impl.nullpm.NullStorageManager;
import org.apache.activemq.artemis.core.server.impl.RoutingContextImpl;
import org.apache.activemq.artemis.core.settings.HierarchicalRepository;
@@ -36,6 +40,7 @@ import org.apache.activemq.artemis.core.settings.impl.AddressSettings;
import org.apache.activemq.artemis.core.settings.impl.HierarchicalObjectRepository;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.utils.RandomUtil;
+import org.apache.activemq.artemis.utils.actors.OrderedExecutorFactory;
import org.apache.activemq.artemis.utils.collections.LinkedList;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -53,9 +58,14 @@ public class PagingManagerImplTest extends ActiveMQTestBase {
HierarchicalRepository<AddressSettings> addressSettings = new HierarchicalObjectRepository<>();
addressSettings.setDefault(new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE));
- final StorageManager storageManager = new NullStorageManager();
+ OrderedExecutorFactory orderedExecutorFactory = getOrderedExecutor();
- PagingStoreFactoryNIO storeFactory = new PagingStoreFactoryNIO(storageManager, getPageDirFile(), 100, null, getOrderedExecutor(), getOrderedExecutor(), true, null);
+ final StorageManager storageManager = new NullStorageManager().setContextSupplier(() -> OperationContextImpl.getContext(orderedExecutorFactory));
+
+ ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
+ runAfter(scheduledExecutorService::shutdownNow);
+
+ PagingStoreFactoryNIO storeFactory = new PagingStoreFactoryNIO(storageManager, getPageDirFile(), 100, scheduledExecutorService, orderedExecutorFactory, true, null);
PagingManagerImpl managerImpl = new PagingManagerImpl(storeFactory, addressSettings);
@@ -73,8 +83,9 @@ public class PagingManagerImplTest extends ActiveMQTestBase {
store.startPaging();
assertTrue(store.page(msg, ctx.getTransaction(), ctx.getContextListing(store.getStoreName())));
+ syncOperationContext();
- Page page = store.depage();
+ Page page = depageOnExecutor(store);
page.open(true);
@@ -88,7 +99,7 @@ public class PagingManagerImplTest extends ActiveMQTestBase {
assertTrue(store.isPaging());
- assertNull(store.depage());
+ assertNull(depageOnExecutor(store));
final RoutingContextImpl ctx2 = new RoutingContextImpl(null);
assertFalse(store.page(msg, ctx2.getTransaction(), ctx2.getContextListing(store.getStoreName())));
@@ -126,4 +137,12 @@ public class PagingManagerImplTest extends ActiveMQTestBase {
}
return buffer;
}
+
+
+ /** depage is now done within the page's executor.
+ * This unit test needs to call the depage within that executor */
+ protected Page depageOnExecutor(final PagingStore store) throws Exception {
+ return callOnExecutor(store.getExecutor(), () -> store.depage(), 10, TimeUnit.SECONDS);
+ }
+
}
diff --git a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PagingStoreImplTest.java b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PagingStoreImplTest.java
index 86062f085f..01cb703dc1 100644
--- a/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PagingStoreImplTest.java
+++ b/tests/unit-tests/src/test/java/org/apache/activemq/artemis/tests/unit/core/paging/impl/PagingStoreImplTest.java
@@ -16,21 +16,14 @@
*/
package org.apache.activemq.artemis.tests.unit.core.paging.impl;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertFalse;
-import static org.junit.jupiter.api.Assertions.assertNull;
-import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertTrue;
-
import java.io.File;
+import java.lang.invoke.MethodHandles;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
-import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -61,6 +54,7 @@ import org.apache.activemq.artemis.core.paging.impl.PageReadWriter;
import org.apache.activemq.artemis.core.paging.impl.PageTransactionInfoImpl;
import org.apache.activemq.artemis.core.paging.impl.PagingStoreImpl;
import org.apache.activemq.artemis.core.paging.impl.PagingStoreTestAccessor;
+import org.apache.activemq.artemis.core.persistence.OperationContext;
import org.apache.activemq.artemis.core.persistence.StorageManager;
import org.apache.activemq.artemis.core.persistence.impl.journal.OperationContextImpl;
import org.apache.activemq.artemis.core.persistence.impl.nullpm.NullStorageManager;
@@ -77,13 +71,12 @@ import org.apache.activemq.artemis.tests.unit.core.postoffice.impl.fakes.FakeQue
import org.apache.activemq.artemis.tests.unit.util.FakePagingManager;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.utils.ActiveMQThreadFactory;
-import org.apache.activemq.artemis.utils.ExecutorFactory;
import org.apache.activemq.artemis.utils.RandomUtil;
import org.apache.activemq.artemis.utils.ReusableLatch;
import org.apache.activemq.artemis.utils.Wait;
import org.apache.activemq.artemis.utils.actors.ArtemisExecutor;
+import org.apache.activemq.artemis.utils.actors.OrderedExecutorFactory;
import org.apache.activemq.artemis.utils.collections.LinkedList;
-import org.apache.activemq.artemis.utils.collections.LinkedListIterator;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -91,7 +84,12 @@ import org.junit.jupiter.api.Timeout;
import org.mockito.Mockito;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.lang.invoke.MethodHandles;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
public class PagingStoreImplTest extends ActiveMQTestBase {
private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());
@@ -104,6 +102,10 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
private static final SimpleString destinationTestName = SimpleString.of("test");
protected ExecutorService executor;
+ protected OrderedExecutorFactory orderedExecutorFactory;
+ protected ScheduledExecutorService scheduledExecutorService;
+ protected OperationContext context;
+ protected NullStorageManager nullStorageManager;
@Test
public void testAddAndRemoveMessages() {
@@ -139,7 +141,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
public void testDoubleStart() throws Exception {
SequentialFileFactory factory = new FakeSequentialFileFactory();
- PagingStore storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), createStorageManagerMock(), factory, new FakeStoreFactory(factory), PagingStoreImplTest.destinationTestName, new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE), getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ PagingStore storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100, createMockManager(), nullStorageManager, factory, new FakeStoreFactory(factory), PagingStoreImplTest.destinationTestName, new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE), orderedExecutorFactory.getExecutor(), true);
storeImpl.start();
@@ -152,12 +154,6 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
}
- @Test
- public void testPageWithNIO() throws Exception {
- ActiveMQTestBase.recreateDirectory(getTestDir());
- testConcurrentPaging(new NIOSequentialFileFactory(new File(getTestDir()), 1).setDatasync(false), 1);
- }
-
@Test
public void testStore() throws Exception {
SequentialFileFactory factory = new FakeSequentialFileFactory();
@@ -166,7 +162,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
AddressSettings addressSettings = new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
- PagingStore storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), createStorageManagerMock(), factory, storeFactory, PagingStoreImplTest.destinationTestName, addressSettings, getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ PagingStore storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100, createMockManager(), nullStorageManager, factory, storeFactory, PagingStoreImplTest.destinationTestName, addressSettings, orderedExecutorFactory.getExecutor(), true);
storeImpl.start();
@@ -189,12 +185,11 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
final RoutingContextImpl ctx = new RoutingContextImpl(null);
assertTrue(storeImpl.page(msg, ctx.getTransaction(), ctx.getContextListing(storeImpl.getStoreName())));
+ syncOperationContext();
assertEquals(1, storeImpl.getNumberOfPages());
- storeImpl.addSyncPoint(OperationContextImpl.getContext());
-
- storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), createStorageManagerMock(), factory, storeFactory, PagingStoreImplTest.destinationTestName, addressSettings, getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100, createMockManager(), nullStorageManager, factory, storeFactory, PagingStoreImplTest.destinationTestName, addressSettings, orderedExecutorFactory.getExecutor(), true);
storeImpl.start();
@@ -211,7 +206,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
PagingStoreFactory storeFactory = new FakeStoreFactory(factory);
- PagingStoreImpl storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), createStorageManagerMock(), factory, storeFactory, PagingStoreImplTest.destinationTestName, new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE), getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ PagingStoreImpl storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100, createMockManager(), nullStorageManager, factory, storeFactory, PagingStoreImplTest.destinationTestName, new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE), orderedExecutorFactory.getExecutor(), true);
storeImpl.start();
@@ -232,24 +227,22 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
Message msg = createMessage(i, storeImpl, destination, buffer);
final RoutingContextImpl ctx = new RoutingContextImpl(null);
assertTrue(storeImpl.page(msg, ctx.getTransaction(), ctx.getContextListing(storeImpl.getStoreName())));
-
+ syncOperationContext();
}
assertEquals(1, storeImpl.getNumberOfPages());
- storeImpl.addSyncPoint(OperationContextImpl.getContext());
-
- Page page = storeImpl.depage();
+ Page page = depageOnExecutor(storeImpl);
page.open(true);
- LinkedList<PagedMessage> msg = page.read(new NullStorageManager());
+ LinkedList<PagedMessage> msg = page.read(nullStorageManager);
assertEquals(numMessages, msg.size());
assertEquals(1, storeImpl.getNumberOfPages());
page.close(false);
- page = storeImpl.depage();
+ page = depageOnExecutor(storeImpl);
assertNull(page);
@@ -275,7 +268,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
PagingStoreFactory storeFactory = new FakeStoreFactory(factory);
- PagingStoreImpl storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), createStorageManagerMock(), factory, storeFactory, PagingStoreImplTest.destinationTestName, new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE), getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ PagingStoreImpl storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 1, createMockManager(), nullStorageManager, factory, storeFactory, PagingStoreImplTest.destinationTestName, new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE), orderedExecutorFactory.getExecutor(), false);
PageSubscription subscription = storeImpl.getCursorProvider().createSubscription(1, null, true);
FakeQueue fakeQueue = new FakeQueue(destination, 1).setDurable(true).setPageSubscription(subscription);
@@ -302,8 +295,9 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
final RoutingContextImpl ctx = new RoutingContextImpl(null);
ctx.addQueue(fakeQueue.getName(), fakeQueue);
assertTrue(storeImpl.page(msg, ctx.getTransaction(), ctx.getContextListing(storeImpl.getStoreName())));
+ syncOperationContext();
if (i > 0 && i % 10 == 0) {
- storeImpl.forceAnotherPage();
+ storeImpl.forceAnotherPage(true);
page++;
assertEquals(page, storeImpl.getNumberOfPages());
}
@@ -393,7 +387,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
PagingStoreFactory storeFactory = new FakeStoreFactory(factory);
- PagingStoreImpl storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), createStorageManagerMock(), factory, storeFactory, PagingStoreImplTest.destinationTestName, new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE), getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ PagingStoreImpl storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 1, createMockManager(), nullStorageManager, factory, storeFactory, PagingStoreImplTest.destinationTestName, new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE), orderedExecutorFactory.getExecutor(), false);
PageSubscription subscription = storeImpl.getCursorProvider().createSubscription(1, null, true);
FakeQueue fakeQueue = new FakeQueue(destination, 1).setDurable(true).setPageSubscription(subscription);
@@ -419,8 +413,9 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
final RoutingContextImpl ctx = new RoutingContextImpl(null);
ctx.addQueue(fakeQueue.getName(), fakeQueue);
assertTrue(storeImpl.page(msg, ctx.getTransaction(), ctx.getContextListing(storeImpl.getStoreName())));
+ syncOperationContext();
if (i > 0 && i % 10 == 0) {
- storeImpl.forceAnotherPage();
+ storeImpl.forceAnotherPage(true);
page++;
}
}
@@ -453,7 +448,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
assertTrue(storeImpl.isPaging());
- storeImpl.forceAnotherPage();
+ storeImpl.forceAnotherPage(true);
assertEquals(11, factory.listFiles("page").size());
@@ -497,9 +492,9 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
PagingStoreFactory storeFactory = new FakeStoreFactory(factory);
- StorageManager storageManager = createStorageManagerMock();
+ StorageManager storageManager = nullStorageManager;
- PagingStoreImpl storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), storageManager, factory, storeFactory, PagingStoreImplTest.destinationTestName, new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE), getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ PagingStoreImpl storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100, createMockManager(), storageManager, factory, storeFactory, PagingStoreImplTest.destinationTestName, new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE), orderedExecutorFactory.getExecutor(), true);
PageSubscription subscription = storeImpl.getCursorProvider().createSubscription(1, null, true);
FakeQueue fakeQueue = new FakeQueue(destination, 1).setDurable(true).setPageSubscription(subscription);
@@ -515,6 +510,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
final RoutingContextImpl ctx = new RoutingContextImpl(null);
ctx.addQueue(fakeQueue.getName(), fakeQueue);
assertTrue(storeImpl.page(msg, ctx.getTransaction(), ctx.getContextListing(storeImpl.getStoreName())));
+ syncOperationContext();
}
assertEquals(1, storeImpl.getNumberOfPages());
@@ -539,12 +535,13 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
@Test
public void testDepageMultiplePages() throws Exception {
+ OperationContextImpl.setContext(context);
SequentialFileFactory factory = new NIOSequentialFileFactory(new File(getPageDir()), 1).setDatasync(false);
SimpleString destination = SimpleString.of("test");
PagingStoreFactory storeFactory = new FakeStoreFactory(factory);
- PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), createStorageManagerMock(), factory, storeFactory, PagingStoreImplTest.destinationTestName, new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE), getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100, createMockManager(), nullStorageManager, factory, storeFactory, PagingStoreImplTest.destinationTestName, new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE), orderedExecutorFactory.getExecutor(), true);
store.start();
@@ -563,29 +560,28 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
buffers.add(buffer);
if (i == 5) {
- store.forceAnotherPage();
+ store.forceAnotherPage(true);
}
Message msg = createMessage(i, store, destination, buffer);
final RoutingContextImpl ctx = new RoutingContextImpl(null);
assertTrue(store.page(msg, ctx.getTransaction(), ctx.getContextListing(store.getStoreName())));
+ syncOperationContext();
}
assertEquals(2, store.getNumberOfPages());
- store.addSyncPoint(OperationContextImpl.getContext());
-
int sequence = 0;
for (int pageNr = 0; pageNr < 2; pageNr++) {
- Page page = store.depage();
+ Page page = depageOnExecutor(store);
logger.debug("numberOfPages = {}", store.getNumberOfPages());
page.open(true);
- LinkedList<PagedMessage> msg = page.read(new NullStorageManager());
+ LinkedList<PagedMessage> msg = page.read(nullStorageManager);
page.close(false, false);
@@ -605,12 +601,13 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
final RoutingContextImpl ctx = new RoutingContextImpl(null);
assertTrue(store.page(msg, ctx.getTransaction(), ctx.getContextListing(store.getStoreName())));
+ syncOperationContext();
- Page newPage = store.depage();
+ Page newPage = depageOnExecutor(store);
newPage.open(true);
- assertEquals(1, newPage.read(new NullStorageManager()).size());
+ assertEquals(1, newPage.read(nullStorageManager).size());
newPage.delete(null);
@@ -618,7 +615,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
assertTrue(store.isPaging());
- assertNull(store.depage());
+ assertNull(depageOnExecutor(store));
assertFalse(store.isPaging());
@@ -632,15 +629,16 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
{
final RoutingContextImpl ctx2 = new RoutingContextImpl(null);
assertTrue(store.page(msg, ctx2.getTransaction(), ctx2.getContextListing(store.getStoreName())));
+ syncOperationContext();
}
- Page page = store.depage();
+ Page page = depageOnExecutor(store);
assertNotNull(page);
page.open(true);
- LinkedList<PagedMessage> msgs = page.read(new NullStorageManager());
+ LinkedList<PagedMessage> msgs = page.read(nullStorageManager);
assertEquals(1, msgs.size());
@@ -652,7 +650,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
assertTrue(store.isPaging());
- assertNull(store.depage());
+ assertNull(depageOnExecutor(store));
assertEquals(0, store.getNumberOfPages());
@@ -661,224 +659,6 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
}
- @Test
- public void testConcurrentDepage() throws Exception {
- SequentialFileFactory factory = new FakeSequentialFileFactory(1, false);
-
- testConcurrentPaging(factory, 10);
- }
-
- protected void testConcurrentPaging(final SequentialFileFactory factory,
- final int numberOfThreads) throws Exception {
- PagingStoreFactory storeFactory = new FakeStoreFactory(factory);
-
- final int MAX_SIZE = 1024 * 10;
-
- final AtomicLong messageIdGenerator = new AtomicLong(0);
-
- final AtomicInteger aliveProducers = new AtomicInteger(numberOfThreads);
-
- final CountDownLatch latchStart = new CountDownLatch(numberOfThreads);
-
- final ConcurrentHashMap<Long, Message> buffers = new ConcurrentHashMap<>();
-
- final ArrayList<Page> readPages = new ArrayList<>();
-
- AddressSettings settings = new AddressSettings().setPageSizeBytes(MAX_SIZE).setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
-
- final PagingStore storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), createStorageManagerMock(), factory, storeFactory, SimpleString.of("test"), settings, getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
-
- storeImpl.start();
-
- assertEquals(0, storeImpl.getNumberOfPages());
-
- // Marked the store to be paged
- storeImpl.startPaging();
-
- assertEquals(1, storeImpl.getNumberOfPages());
-
- final SimpleString destination = SimpleString.of("test");
-
- class WriterThread extends Thread {
-
- Exception e;
-
- @Override
- public void run() {
-
- try {
- boolean firstTime = true;
- while (true) {
- long id = messageIdGenerator.incrementAndGet();
-
- // Each thread will Keep paging until all the messages are depaged.
- // This is possible because the depage thread is not actually reading the pages.
- // Just using the internal API to remove it from the page file system
- Message msg = createMessage(id, storeImpl, destination, createRandomBuffer(id, 5));
- final RoutingContextImpl ctx2 = new RoutingContextImpl(null);
- if (storeImpl.page(msg, ctx2.getTransaction(), ctx2.getContextListing(storeImpl.getStoreName()))) {
- buffers.put(id, msg);
- } else {
- break;
- }
-
- if (firstTime) {
- // We have at least one data paged. So, we can start depaging now
- latchStart.countDown();
- firstTime = false;
- }
- }
- } catch (Exception e1) {
- e1.printStackTrace();
- this.e = e1;
- } finally {
- aliveProducers.decrementAndGet();
- }
- }
- }
-
- final class ReaderThread extends Thread {
-
- Exception e;
-
- @Override
- public void run() {
- try {
- // Wait every producer to produce at least one message
- ActiveMQTestBase.waitForLatch(latchStart);
-
- while (aliveProducers.get() > 0) {
- Page page = storeImpl.depage();
- if (page != null) {
- readPages.add(page);
- }
- }
- } catch (Exception e1) {
- e1.printStackTrace();
- this.e = e1;
- }
- }
- }
-
- WriterThread[] producerThread = new WriterThread[numberOfThreads];
-
- for (int i = 0; i < numberOfThreads; i++) {
- producerThread[i] = new WriterThread();
- producerThread[i].start();
- }
-
- ReaderThread consumer = new ReaderThread();
- consumer.start();
-
- for (int i = 0; i < numberOfThreads; i++) {
- producerThread[i].join();
- if (producerThread[i].e != null) {
- throw producerThread[i].e;
- }
- }
-
- consumer.join();
-
- if (consumer.e != null) {
- throw consumer.e;
- }
-
- final ConcurrentMap<Long, Message> buffers2 = new ConcurrentHashMap<>();
-
- for (Page page : readPages) {
- page.open(true);
- LinkedList<PagedMessage> msgs = page.read(new NullStorageManager());
- page.close(false, false);
-
-
- try (LinkedListIterator<PagedMessage> iter = msgs.iterator()) {
- while (iter.hasNext()) {
- PagedMessage msg = iter.next();
- long id = msg.getMessage().toCore().getBodyBuffer().readLong();
- msg.getMessage().toCore().getBodyBuffer().resetReaderIndex();
-
- Message msgWritten = buffers.remove(id);
- buffers2.put(id, msg.getMessage());
- assertNotNull(msgWritten);
- assertEquals(msg.getMessage().getAddress(), msgWritten.getAddress());
- ActiveMQTestBase.assertEqualsBuffers(10, msgWritten.toCore().getBodyBuffer(), msg.getMessage().toCore().getBodyBuffer());
- }
- }
- }
-
- assertEquals(0, buffers.size());
-
- List<String> files = factory.listFiles("page");
-
- assertTrue(files.size() != 0);
-
- for (String file : files) {
- SequentialFile fileTmp = factory.createSequentialFile(file);
- fileTmp.open();
- assertTrue(fileTmp.size() <= MAX_SIZE, "The page file size (" + fileTmp.size() + ") shouldn't be > " + MAX_SIZE);
- fileTmp.close();
- }
-
- PagingStore storeImpl2 = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), createStorageManagerMock(), factory, storeFactory, SimpleString.of("test"), settings, getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
- storeImpl2.start();
-
- long numberOfPages = storeImpl2.getNumberOfPages();
- assertTrue(numberOfPages != 0);
-
- storeImpl2.startPaging();
-
- storeImpl2.startPaging();
-
- assertEquals(numberOfPages, storeImpl2.getNumberOfPages());
-
- long lastMessageId = messageIdGenerator.incrementAndGet();
- Message lastMsg = createMessage(lastMessageId, storeImpl, destination, createRandomBuffer(lastMessageId, 5));
-
- storeImpl2.forceAnotherPage();
-
- final RoutingContextImpl ctx = new RoutingContextImpl(null);
- storeImpl2.page(lastMsg, ctx.getTransaction(), ctx.getContextListing(storeImpl2.getStoreName()));
- buffers2.put(lastMessageId, lastMsg);
-
- Page lastPage = null;
- while (true) {
- Page page = storeImpl2.depage();
- if (page == null) {
- break;
- }
-
- lastPage = page;
-
- page.open(true);
-
- LinkedList<PagedMessage> msgs = page.read(new NullStorageManager());
-
- page.close(false, false);
-
- msgs.forEach((msg) -> {
- long id = msg.getMessage().toCore().getBodyBuffer().readLong();
- Message msgWritten = buffers2.remove(id);
- assertNotNull(msgWritten);
- assertEquals(msg.getMessage().getAddress(), msgWritten.getAddress());
- ActiveMQTestBase.assertEqualsByteArrays(msgWritten.toCore().getBodyBuffer().writerIndex(), msgWritten.toCore().getBodyBuffer().toByteBuffer().array(), msg.getMessage().toCore().getBodyBuffer().toByteBuffer().array());
- });
- }
-
- lastPage.open(true);
- LinkedList<PagedMessage> lastMessages = lastPage.read(new NullStorageManager());
- lastPage.close(false, false);
- assertEquals(1, lastMessages.size());
-
- lastMessages.get(0).getMessage().toCore().getBodyBuffer().resetReaderIndex();
- assertEquals(lastMessages.get(0).getMessage().toCore().getBodyBuffer().readLong(), lastMessageId);
-
- assertEquals(0, buffers2.size());
-
- assertEquals(0, storeImpl.getAddressSize());
-
- storeImpl.stop();
- }
-
@Test
public void testRestartPage() throws Throwable {
clearDataRecreateServerDirs();
@@ -890,7 +670,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
AddressSettings settings = new AddressSettings().setPageSizeBytes(MAX_SIZE).setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
- final PagingStore storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), createStorageManagerMock(), factory, storeFactory, SimpleString.of("test"), settings, getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ final PagingStore storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100, createMockManager(), nullStorageManager, factory, storeFactory, SimpleString.of("test"), settings, orderedExecutorFactory.getExecutor(), true);
storeImpl.start();
@@ -899,7 +679,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
// Marked the store to be paged
storeImpl.startPaging();
- storeImpl.depage();
+ depageOnExecutor(storeImpl);
assertNull(storeImpl.getCurrentPage());
@@ -913,6 +693,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
@Test
public void testOrderOnPaging() throws Throwable {
ExecutorService executorService = Executors.newFixedThreadPool(2);
+ runAfter(executorService::shutdownNow);
try {
clearDataRecreateServerDirs();
SequentialFileFactory factory = new FakeSequentialFileFactory();
@@ -923,9 +704,10 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
AddressSettings settings = new AddressSettings().setPageSizeBytes(MAX_SIZE).setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
- final PagingStore store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), createStorageManagerMock(), factory, storeFactory, SimpleString.of("test"), settings, getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), false);
+ final PagingStore store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100, createMockManager(), nullStorageManager, factory, storeFactory, SimpleString.of("test"), settings, orderedExecutorFactory.getExecutor(), false);
store.start();
+ runAfter(store::stop);
assertEquals(0, store.getNumberOfPages());
@@ -955,6 +737,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
final RoutingContextImpl ctx2 = new RoutingContextImpl(null);
store.page(msg, ctx2.getTransaction(), ctx2.getContextListing(store.getStoreName()));
}
+ syncOperationContext();
} catch (Throwable e) {
e.printStackTrace();
errors.add(e);
@@ -971,11 +754,11 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
AtomicInteger msgsRead = new AtomicInteger(0);
while (msgsRead.get() < NUMBER_OF_MESSAGES) {
- Page page = store.depage();
+ Page page = depageOnExecutor(store);
AtomicInteger countOnPage = new AtomicInteger(0);
if (page != null) {
page.open(true);
- LinkedList<PagedMessage> messages = page.read(new NullStorageManager());
+ LinkedList<PagedMessage> messages = page.read(nullStorageManager);
messages.forEach(pgmsg -> {
assertEquals(countOnPage.getAndIncrement(), pgmsg.getMessageNumber());
Message msg = pgmsg.getMessage();
@@ -1020,7 +803,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
AddressSettings settings = new AddressSettings().setPageSizeBytes(MAX_SIZE).setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE);
- final PagingStore storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100, createMockManager(), createStorageManagerMock(), factory, storeFactory, SimpleString.of("test"), settings, getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ final PagingStore storeImpl = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100, createMockManager(), nullStorageManager, factory, storeFactory, SimpleString.of("test"), settings, orderedExecutorFactory.getExecutor(), true);
storeImpl.start();
@@ -1061,10 +844,10 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
AtomicLong msgsRead = new AtomicLong(0);
while (msgsRead.get() < num1 + num2) {
- page = storeImpl.depage();
+ page = depageOnExecutor(storeImpl);
assertNotNull(page, "no page after read " + msgsRead + " msg");
page.open(true);
- LinkedList<PagedMessage> messages = page.read(new NullStorageManager());
+ LinkedList<PagedMessage> messages = page.read(nullStorageManager);
messages.forEach(pgmsg -> {
Message msg = pgmsg.getMessage();
@@ -1084,12 +867,12 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
PagingStoreFactory storeFactory = new FakeStoreFactory(factory);
- PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100,
- createMockManager(), createStorageManagerMock(), factory, storeFactory,
+ PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100,
+ createMockManager(), nullStorageManager, factory, storeFactory,
PagingStoreImplTest.destinationTestName,
new AddressSettings()
.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE),
- getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ orderedExecutorFactory.getExecutor(), true);
store.start();
try (AssertionLoggerHandler loggerHandler = new AssertionLoggerHandler()) {
@@ -1105,12 +888,12 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
PagingStoreFactory storeFactory = new FakeStoreFactory(factory);
- PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100,
- createMockManager(), createStorageManagerMock(), factory, storeFactory,
+ PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100,
+ createMockManager(), nullStorageManager, factory, storeFactory,
PagingStoreImplTest.destinationTestName,
new AddressSettings()
.setAddressFullMessagePolicy(AddressFullMessagePolicy.PAGE),
- getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ orderedExecutorFactory.getExecutor(), true);
store.start();
try (AssertionLoggerHandler loggerHandler = new AssertionLoggerHandler()) {
store.startPaging();
@@ -1125,12 +908,12 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
PagingStoreFactory storeFactory = new FakeStoreFactory(factory);
- PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100,
- createMockManager(), createStorageManagerMock(), factory, storeFactory,
+ PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100,
+ createMockManager(), nullStorageManager, factory, storeFactory,
PagingStoreImplTest.destinationTestName,
new AddressSettings()
.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK),
- getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ orderedExecutorFactory.getExecutor(), true);
store.start();
try {
@@ -1196,12 +979,12 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
}
};
- PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100,
- pagingManager, createStorageManagerMock(), factory, storeFactory,
+ PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100,
+ pagingManager, nullStorageManager, factory, storeFactory,
PagingStoreImplTest.destinationTestName,
new AddressSettings()
.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK),
- getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ orderedExecutorFactory.getExecutor(), true);
store.start();
try {
@@ -1240,12 +1023,12 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
PagingStoreFactory storeFactory = new FakeStoreFactory(factory);
- PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100,
- createMockManager(), createStorageManagerMock(), factory, storeFactory,
+ PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100,
+ createMockManager(), nullStorageManager, factory, storeFactory,
PagingStoreImplTest.destinationTestName,
new AddressSettings()
.setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK),
- getExecutorFactory().getExecutor(), getExecutorFactory().getExecutor(), true);
+ orderedExecutorFactory.getExecutor(), true);
store.start();
try {
@@ -1319,14 +1102,6 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
return new FakePagingManager();
}
- private StorageManager createStorageManagerMock() {
- return new NullStorageManager();
- }
-
- private ExecutorFactory getExecutorFactory() {
- return () -> ArtemisExecutor.delegate(executor);
- }
-
protected void writePageMessage(final PagingStore storeImpl,
final long id) throws Exception {
Message msg = createMessage(id, storeImpl, PagingStoreImplTest.destinationTestName, createRandomBuffer(id, 10));
@@ -1334,6 +1109,7 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
final RoutingContextImpl ctx2 = new RoutingContextImpl(null);
storeImpl.page(msg, ctx2.getTransaction(), ctx2.getContextListing(storeImpl.getStoreName()));
+ syncOperationContext();
}
private CoreMessage createMessage(final long id,
@@ -1363,12 +1139,19 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
public void setUp() throws Exception {
super.setUp();
executor = Executors.newSingleThreadExecutor(ActiveMQThreadFactory.defaultThreadFactory(getClass().getName()));
+ orderedExecutorFactory = new OrderedExecutorFactory(executor);
+ scheduledExecutorService = Executors.newScheduledThreadPool(10);
+ OperationContextImpl.clearContext();
+ context = OperationContextImpl.getContext(orderedExecutorFactory);
+ runAfter(OperationContextImpl::clearContext);
+ nullStorageManager = new NullStorageManager().setContextSupplier(() -> OperationContextImpl.getContext(orderedExecutorFactory));
}
@Override
@AfterEach
public void tearDown() throws Exception {
executor.shutdown();
+ scheduledExecutorService.shutdownNow();
super.tearDown();
}
@@ -1461,11 +1244,11 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
PagingManager mockManager = Mockito.mock(PagingManager.class);
ArtemisExecutor sameThreadExecutor = Runnable::run;
- PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, null, 100,
- mockManager, createStorageManagerMock(), factory, storeFactory,
+ PagingStoreImpl store = new PagingStoreImpl(PagingStoreImplTest.destinationTestName, scheduledExecutorService, 100,
+ mockManager, nullStorageManager, factory, storeFactory,
PagingStoreImplTest.destinationTestName,
new AddressSettings().setAddressFullMessagePolicy(AddressFullMessagePolicy.BLOCK),
- sameThreadExecutor, sameThreadExecutor, true);
+ sameThreadExecutor, true);
store.start();
try {
@@ -1495,4 +1278,11 @@ public class PagingStoreImplTest extends ActiveMQTestBase {
store.stop();
}
}
+
+ /** depage is done within the page's executor.
+ * This unit test needs to call the depage within that executor */
+ private Page depageOnExecutor(final PagingStore store) throws Exception {
+ return callOnExecutor(store.getExecutor(), () -> store.depage(), 10, TimeUnit.SECONDS);
+ }
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@activemq.apache.org
For additional commands, e-mail: commits-help@activemq.apache.org
For further information, visit: https://activemq.apache.org/contact