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

This closes #2942

This commit is contained in:
Clebert Suconic 2020-01-21 12:14:03 -05:00
parent fee21f7bf6 c3e74e055c
commit 5dbce95431
1 changed files with 65 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -310,10 +310,40 @@ public final class DescribeJournal {
final StringBuffer bufferFailingTransactions = new StringBuffer(); final StringBuffer bufferFailingTransactions = new StringBuffer();
int messageCount = 0; final class Count {
Map<Long, Integer> messageRefCounts = new HashMap<>();
int preparedMessageCount = 0; int value;
Map<Long, Integer> preparedMessageRefCount = new HashMap<>();
Count(int v) {
value = v;
}
@Override
public String toString() {
return Integer.toString(value);
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
Count count = (Count) o;
return value == count.value;
}
@Override
public int hashCode() {
return Integer.hashCode(value);
}
}
long messageCount = 0;
long largeMessageCount = 0;
Map<Long, Count> messageRefCounts = new HashMap<>();
long preparedMessageCount = 0;
long preparedLargeMessageCount = 0;
Map<Long, Count> preparedMessageRefCount = new HashMap<>();
journal.load(records, preparedTransactions, new TransactionFailureCallback() { journal.load(records, preparedTransactions, new TransactionFailureCallback() {
@Override @Override
@ -337,26 +367,28 @@ public final class DescribeJournal {
long queueIDForCounter = 0; long queueIDForCounter = 0;
Object o = newObjectEncoding(info); Object o = newObjectEncoding(info);
if (info.getUserRecordType() == JournalRecordIds.ADD_MESSAGE) { final byte userRecordType = info.getUserRecordType();
if (userRecordType == ADD_MESSAGE || userRecordType == ADD_MESSAGE_PROTOCOL) {
messageCount++; messageCount++;
} else if (info.getUserRecordType() == JournalRecordIds.ADD_REF) { } else if (userRecordType == ADD_LARGE_MESSAGE) {
largeMessageCount++;
} else if (userRecordType == JournalRecordIds.ADD_REF) {
ReferenceDescribe ref = (ReferenceDescribe) o; ReferenceDescribe ref = (ReferenceDescribe) o;
Integer count = messageRefCounts.get(ref.refEncoding.queueID); Count count = messageRefCounts.get(ref.refEncoding.queueID);
if (count == null) { if (count == null) {
count = 1; count = new Count(0);
messageRefCounts.put(ref.refEncoding.queueID, count); messageRefCounts.put(ref.refEncoding.queueID, count);
} else {
messageRefCounts.put(ref.refEncoding.queueID, count + 1);
} }
} else if (info.getUserRecordType() == JournalRecordIds.ACKNOWLEDGE_REF) { count.value++;
} else if (userRecordType == JournalRecordIds.ACKNOWLEDGE_REF) {
AckDescribe ref = (AckDescribe) o; AckDescribe ref = (AckDescribe) o;
Integer count = messageRefCounts.get(ref.refEncoding.queueID); Count count = messageRefCounts.get(ref.refEncoding.queueID);
if (count == null) { if (count == null) {
messageRefCounts.put(ref.refEncoding.queueID, 0); messageRefCounts.put(ref.refEncoding.queueID, new Count(0));
} else { } else {
messageRefCounts.put(ref.refEncoding.queueID, count - 1); count.value--;
} }
} else if (info.getUserRecordType() == JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE) { } else if (userRecordType == JournalRecordIds.PAGE_CURSOR_COUNTER_VALUE) {
PageCountRecord encoding = (PageCountRecord) o; PageCountRecord encoding = (PageCountRecord) o;
queueIDForCounter = encoding.getQueueID(); queueIDForCounter = encoding.getQueueID();
@ -364,7 +396,7 @@ public final class DescribeJournal {
subsCounter.loadValue(info.id, encoding.getValue(), encoding.getPersistentSize()); subsCounter.loadValue(info.id, encoding.getValue(), encoding.getPersistentSize());
subsCounter.processReload(); subsCounter.processReload();
} else if (info.getUserRecordType() == JournalRecordIds.PAGE_CURSOR_COUNTER_INC) { } else if (userRecordType == JournalRecordIds.PAGE_CURSOR_COUNTER_INC) {
PageCountRecordInc encoding = (PageCountRecordInc) o; PageCountRecordInc encoding = (PageCountRecordInc) o;
queueIDForCounter = encoding.getQueueID(); queueIDForCounter = encoding.getQueueID();
@ -395,17 +427,19 @@ public final class DescribeJournal {
for (RecordInfo info : tx.getRecords()) { for (RecordInfo info : tx.getRecords()) {
Object o = newObjectEncoding(info); Object o = newObjectEncoding(info);
out.println("- " + describeRecord(info, o, safe)); out.println("- " + describeRecord(info, o, safe));
if (info.getUserRecordType() == 31) { final byte userRecordType = info.getUserRecordType();
if (userRecordType == ADD_MESSAGE || userRecordType == ADD_MESSAGE_PROTOCOL) {
preparedMessageCount++; preparedMessageCount++;
} else if (info.getUserRecordType() == 32) { } else if (userRecordType == ADD_LARGE_MESSAGE) {
preparedLargeMessageCount++;
} else if (userRecordType == ADD_REF) {
ReferenceDescribe ref = (ReferenceDescribe) o; ReferenceDescribe ref = (ReferenceDescribe) o;
Integer count = preparedMessageRefCount.get(ref.refEncoding.queueID); Count count = preparedMessageRefCount.get(ref.refEncoding.queueID);
if (count == null) { if (count == null) {
count = 1; count = new Count(0);
preparedMessageRefCount.put(ref.refEncoding.queueID, count); preparedMessageRefCount.put(ref.refEncoding.queueID, count);
} else {
preparedMessageRefCount.put(ref.refEncoding.queueID, count + 1);
} }
count.value++;
} }
} }
@ -425,16 +459,17 @@ public final class DescribeJournal {
out.println("### Message Counts ###"); out.println("### Message Counts ###");
out.println("message count=" + messageCount); out.println("message count=" + messageCount);
out.println("large message count=" + largeMessageCount);
out.println("message reference count"); out.println("message reference count");
for (Map.Entry<Long, Integer> longIntegerEntry : messageRefCounts.entrySet()) { messageRefCounts.forEach((queueId, count) -> {
out.println("queue id " + longIntegerEntry.getKey() + ",count=" + longIntegerEntry.getValue()); out.println("queue id " + queueId + ",count=" + count);
} });
out.println("prepared message count=" + preparedMessageCount); out.println("prepared message count=" + preparedMessageCount);
out.println("prepared large message count=" + preparedLargeMessageCount);
for (Map.Entry<Long, Integer> longIntegerEntry : preparedMessageRefCount.entrySet()) { out.println("prepared message reference count");
out.println("queue id " + longIntegerEntry.getKey() + ",count=" + longIntegerEntry.getValue()); preparedMessageRefCount.forEach((queueId, count) -> {
} out.println("queue id " + queueId + ",count=" + count);
});
journal.stop(); journal.stop();