Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-1287/ARTEMIS-1292 Complete Page on the Journal 

fixing PagingTest.testDeletePhysicalPages
This commit is contained in:
Clebert Suconic 2017-07-14 15:04:43 -04:00
parent baeca47b44
commit f18b4ee0c9
7 changed files with 36 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
artemis-core-client/src/main/java/org/apache/activemq/artemis/core/message/impl/CoreMessage.java
View File

@ -1079,10 +1079,13 @@ public class CoreMessage extends RefCountMessage implements ICoreMessage {
@Override
public String toString() {
try {
checkProperties();
return "CoreMessage[messageID=" + messageID + ",durable=" + isDurable() + ",userID=" + getUserID() + ",priority=" + this.getPriority() +
", timestamp=" + toDate(getTimestamp()) + ",expiration=" + toDate(getExpiration()) +
", durable=" + durable + ", address=" + getAddress() + ",properties=" + properties.toString() + "]@" + System.identityHashCode(this);
", durable=" + durable + ", address=" + getAddress() + ",properties=" + properties + "]@" + System.identityHashCode(this);
} catch (Throwable e) {
e.printStackTrace();
System.exit(-1);
return "ServerMessage[messageID=" + messageID + "]";
}
}

2
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/PageSubscription.java
View File

@ -96,7 +96,7 @@ public interface PageSubscription {
*/
void reloadACK(PagePosition position);
void reloadPageCompletion(PagePosition position) throws Exception;
boolean reloadPageCompletion(PagePosition position) throws Exception;
void reloadPageInfo(long pageNr);

7
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/cursor/impl/PageSubscriptionImpl.java
View File

@ -190,7 +190,10 @@ final class PageSubscriptionImpl implements PageSubscription {
* cursor/subscription.
*/
@Override
public void reloadPageCompletion(PagePosition position) throws Exception {
public boolean reloadPageCompletion(PagePosition position) throws Exception {
if (!pageStore.checkPageFileExists((int)position.getPageNr())) {
return false;
}
// if the current page is complete, we must move it out of the way
if (pageStore != null && pageStore.getCurrentPage() != null &&
pageStore.getCurrentPage().getPageId() == position.getPageNr()) {
@ -201,6 +204,8 @@ final class PageSubscriptionImpl implements PageSubscription {
synchronized (consumedPages) {
consumedPages.put(Long.valueOf(position.getPageNr()), info);
}
return true;
}
@Override

16
artemis-server/src/main/java/org/apache/activemq/artemis/core/paging/impl/PagingStoreImpl.java
View File

@ -520,6 +520,12 @@ public class PagingStoreImpl implements PagingStore {
@Override
public boolean checkPageFileExists(final int pageNumber) {
String fileName = createFileName(pageNumber);
try {
checkFileFactory();
} catch (Exception ignored) {
}
SequentialFile file = fileFactory.createSequentialFile(fileName);
return file.exists();
}
@ -528,9 +534,7 @@ public class PagingStoreImpl implements PagingStore {
public Page createPage(final int pageNumber) throws Exception {
String fileName = createFileName(pageNumber);
if (fileFactory == null) {
fileFactory = storeFactory.newFileFactory(getStoreName());
}
checkFileFactory();
SequentialFile file = fileFactory.createSequentialFile(fileName);
@ -546,6 +550,12 @@ public class PagingStoreImpl implements PagingStore {
return page;
}
private void checkFileFactory() throws Exception {
if (fileFactory == null) {
fileFactory = storeFactory.newFileFactory(getStoreName());
}
}
@Override
public void forceAnotherPage() throws Exception {
openNewPage();

7
artemis-server/src/main/java/org/apache/activemq/artemis/core/persistence/impl/journal/AbstractJournalStorageManager.java
View File

@ -1085,7 +1085,12 @@ public abstract class AbstractJournalStorageManager implements StorageManager {
PageSubscription sub = locateSubscription(encoding.queueID, pageSubscriptions, queueInfos, pagingManager);
if (sub != null) {
sub.reloadPageCompletion(encoding.position);
if (!sub.reloadPageCompletion(encoding.position)) {
if (logger.isDebugEnabled()) {
logger.debug("Complete page " + encoding.position.getPageNr() + " doesn't exist on page manager " + sub.getPagingStore().getAddress());
}
messageJournal.appendDeleteRecord(record.id, false);
}
} else {
ActiveMQServerLogger.LOGGER.cantFindQueueOnPageComplete(encoding.queueID);
messageJournal.appendDeleteRecord(record.id, false);

2
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/client/HangConsumerTest.java
View File

@ -71,6 +71,7 @@ import org.apache.activemq.artemis.spi.core.security.ActiveMQJAASSecurityManager
import org.apache.activemq.artemis.spi.core.security.ActiveMQSecurityManager;
import org.apache.activemq.artemis.spi.core.security.jaas.InVMLoginModule;
import org.apache.activemq.artemis.tests.util.ActiveMQTestBase;
import org.apache.activemq.artemis.tests.util.Wait;
import org.apache.activemq.artemis.utils.ExecutorFactory;
import org.apache.activemq.artemis.utils.ReusableLatch;
import org.junit.Assert;
@ -151,6 +152,7 @@ public class HangConsumerTest extends ActiveMQTestBase {
// a flush to guarantee any pending task is finished on flushing out delivery and pending msgs
queue.flushExecutor();
Wait.waitFor(() -> getMessageCount(queue) == 2);
Assert.assertEquals(2, getMessageCount(queue));
Assert.assertEquals(2, getMessagesAdded(queue));

7
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/paging/PagingTest.java
View File

@ -1430,14 +1430,14 @@ public class PagingTest extends ActiveMQTestBase {
producer = session.createProducer(PagingTest.ADDRESS);
for (int i = 0; i < numberOfMessages; i++) {
for (int i = 0; i < numberOfMessages * 2; i++) {
message = session.createMessage(true);
ActiveMQBuffer bodyLocal = message.getBodyBuffer();
bodyLocal.writeBytes(body);
message.putIntProperty(new SimpleString("id"), i);
message.putIntProperty(new SimpleString("theid"), i);
producer.send(message);
if (i % 1000 == 0) {
@ -1466,12 +1466,13 @@ public class PagingTest extends ActiveMQTestBase {
for (int msgCount = 0; msgCount < numberOfMessages; msgCount++) {
log.info("Received " + msgCount);
msgReceived++;
ClientMessage msg = consumer.receiveImmediate();
ClientMessage msg = consumer.receive(5000);
if (msg == null) {
log.info("It's null. leaving now");
sessionConsumer.commit();
fail("Didn't receive a message");
}
System.out.println("Message " + msg.getIntProperty(SimpleString.toSimpleString("theid")));
msg.acknowledge();
if (msgCount % 5 == 0) {