diff --git a/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java b/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java index a0eb10b6fa..c146135c4e 100644 --- a/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java +++ b/activemq-core/src/main/java/org/apache/activemq/store/kahadb/MessageDatabase.java @@ -445,6 +445,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe } // public for testing + @SuppressWarnings("rawtypes") public Location getFirstInProgressTxLocation() { Location l = null; synchronized (inflightTransactions) { @@ -854,7 +855,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe } } - if (!checkpointThread.isAlive()) { + if (checkpointThread == null || !checkpointThread.isAlive()) { startCheckpoint(); } return location; @@ -967,6 +968,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe }); } + @SuppressWarnings("rawtypes") protected void process(final KahaAddMessageCommand command, final Location location) throws IOException { if (command.hasTransactionInfo()) { List inflightTx = getInflightTx(command.getTransactionInfo(), location); @@ -985,6 +987,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe } } + @SuppressWarnings("rawtypes") protected void process(final KahaRemoveMessageCommand command, final Location location) throws IOException { if (command.hasTransactionInfo()) { List inflightTx = getInflightTx(command.getTransactionInfo(), location); @@ -1047,6 +1050,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe } } + @SuppressWarnings("rawtypes") protected void process(KahaCommitCommand command, Location location, final Runnable after) throws IOException { TransactionId key = TransactionIdConversion.convert(command.getTransactionInfo()); List inflightTx; @@ -1077,6 +1081,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe } } + @SuppressWarnings("rawtypes") protected void process(KahaPrepareCommand command, Location location) { TransactionId key = TransactionIdConversion.convert(command.getTransactionInfo()); synchronized (inflightTransactions) { @@ -1087,6 +1092,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe } } + @SuppressWarnings("rawtypes") protected void process(KahaRollbackCommand command, Location location) throws IOException { TransactionId key = TransactionIdConversion.convert(command.getTransactionInfo()); List updates = null; @@ -1101,6 +1107,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe } } + @SuppressWarnings("rawtypes") private void persistRedeliveryCount(List updates) throws IOException { if (updates != null) { for (Operation operation : updates) { @@ -1557,7 +1564,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe if (dataIn.readBoolean()) { value.subscriptions = new BTreeIndex(pageFile, dataIn.readLong()); value.subscriptionAcks = new BTreeIndex(pageFile, dataIn.readLong()); - if (metadata.version >= 3) { + if (metadata.version >= VERSION) { value.ackPositions = new ListIndex(pageFile, dataIn.readLong()); } else { // upgrade @@ -1963,7 +1970,9 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe // ///////////////////////////////////////////////////////////////// // Transaction related implementation methods. // ///////////////////////////////////////////////////////////////// + @SuppressWarnings("rawtypes") private final LinkedHashMap> inflightTransactions = new LinkedHashMap>(); + @SuppressWarnings("rawtypes") protected final LinkedHashMap> preparedTransactions = new LinkedHashMap>(); protected final Set ackedAndPrepared = new HashSet(); @@ -1994,6 +2003,7 @@ public abstract class MessageDatabase extends ServiceSupport implements BrokerSe } } + @SuppressWarnings("rawtypes") private List getInflightTx(KahaTransactionInfo info, Location location) { TransactionId key = TransactionIdConversion.convert(info); List tx;