You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by or...@apache.org on 2019/04/08 08:45:42 UTC
[qpid-broker-j] branch 7.1.x updated (99b6cc9 -> 391954f)
This is an automated email from the ASF dual-hosted git repository.
orudyy pushed a change to branch 7.1.x
in repository https://gitbox.apache.org/repos/asf/qpid-broker-j.git.
from 99b6cc9 [maven-release-plugin] prepare for next development iteration
new c3ae728 QPID-8294: [Broker-J][Oracle Message Store] Batch delete fails for more than 1000 messages
new 391954f QPID-8294: [Broker-J] Fix code formatting
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../store/jdbc/AbstractJDBCMessageStore.java | 69 +++++++++++++---------
.../server/store/jdbc/JDBCMessageStoreTest.java | 21 +++++++
2 files changed, 63 insertions(+), 27 deletions(-)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-broker-j] 02/02: QPID-8294: [Broker-J] Fix code formatting
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orudyy pushed a commit to branch 7.1.x
in repository https://gitbox.apache.org/repos/asf/qpid-broker-j.git
commit 391954f59e442872146351302007a823bd266574
Author: Alex Rudyy <or...@apache.org>
AuthorDate: Mon Apr 8 09:19:41 2019 +0100
QPID-8294: [Broker-J] Fix code formatting
(cherry picked from commit 920db7be6d8248b1662044000a7bf809605ca26c)
---
.../store/jdbc/AbstractJDBCMessageStore.java | 29 +++++++++++++---------
1 file changed, 17 insertions(+), 12 deletions(-)
diff --git a/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
index f3ae9ad..171a2f9 100644
--- a/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
+++ b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
@@ -462,28 +462,29 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
void removeMessages(List<Long> messageIds)
{
- if(messageIds != null && !messageIds.isEmpty())
+ if (messageIds != null && !messageIds.isEmpty())
{
- try(Connection conn = newConnection())
+ try (Connection conn = newConnection())
{
try
{
- for (int i = 0; i <= messageIds.size() / IN_CLAUSE_MAX_SIZE; i++) {
+ for (int i = 0; i <= messageIds.size() / IN_CLAUSE_MAX_SIZE; i++)
+ {
List<Long> boundMessageIds = messageIds.stream()
- .skip(i * IN_CLAUSE_MAX_SIZE)
- .limit(IN_CLAUSE_MAX_SIZE)
- .collect(Collectors.toList());
+ .skip(i * IN_CLAUSE_MAX_SIZE)
+ .limit(IN_CLAUSE_MAX_SIZE)
+ .collect(Collectors.toList());
removeMessagesFromDatabase(conn, boundMessageIds);
}
}
- catch(SQLException e)
+ catch (SQLException e)
{
try
{
conn.rollback();
}
- catch(SQLException t)
+ catch (SQLException t)
{
// ignore - we are re-throwing underlying exception
}
@@ -493,15 +494,19 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
}
catch (SQLException e)
{
- throw new StoreException("Error removing messages with ids " + messageIds + " from database: " + e.getMessage(), e);
+ throw new StoreException("Error removing messages with ids "
+ + messageIds
+ + " from database: "
+ + e.getMessage(), e);
}
}
}
- void removeMessagesFromDatabase(Connection conn, List<Long> messageIds) throws SQLException {
+ void removeMessagesFromDatabase(Connection conn, List<Long> messageIds) throws SQLException
+ {
String inpart = messageIds.stream().map(Object::toString).collect(Collectors.joining(", ", "(", ")"));
- try(Statement stmt = conn.createStatement())
+ try (Statement stmt = conn.createStatement())
{
int results = stmt.executeUpdate("DELETE FROM " + getMetaDataTableName() + " WHERE message_id IN " + inpart);
stmt.close();
@@ -515,7 +520,7 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
getLogger().debug("Deleted metadata for messages {}", messageIds);
}
- try(Statement stmt = conn.createStatement())
+ try (Statement stmt = conn.createStatement())
{
stmt.executeUpdate("DELETE FROM " + getMessageContentTableName() + " WHERE message_id IN " + inpart);
getLogger().debug("Deleted content for messages {}", messageIds);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org
[qpid-broker-j] 01/02: QPID-8294: [Broker-J][Oracle Message Store]
Batch delete fails for more than 1000 messages
Posted by or...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
orudyy pushed a commit to branch 7.1.x
in repository https://gitbox.apache.org/repos/asf/qpid-broker-j.git
commit c3ae728743a0ec9ac6d4960c9c0a7d49e4f122a3
Author: overmeulen <ov...@murex.com>
AuthorDate: Thu Apr 4 15:14:57 2019 +0200
QPID-8294: [Broker-J][Oracle Message Store] Batch delete fails for more than 1000 messages
This closes #23
(cherry picked from commit ba933e93bc86f73ec2cf97d4182069bbcc69d0d3)
---
.../store/jdbc/AbstractJDBCMessageStore.java | 56 +++++++++++++---------
.../server/store/jdbc/JDBCMessageStoreTest.java | 21 ++++++++
2 files changed, 54 insertions(+), 23 deletions(-)
diff --git a/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
index cc0e3c0..f3ae9ad 100644
--- a/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
+++ b/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/AbstractJDBCMessageStore.java
@@ -82,6 +82,8 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
private static final String XID_TABLE_NAME_SUFFIX = "QPID_XIDS";
private static final String XID_ACTIONS_TABLE_NAME_SUFFIX = "QPID_XID_ACTIONS";
+ private static final int IN_CLAUSE_MAX_SIZE = Integer.getInteger("qpid.jdbcstore.inClauseMaxSize",1000);
+
private static final int DB_VERSION = 8;
private final AtomicLong _messageId = new AtomicLong(0);
@@ -458,38 +460,22 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
return _messageRemovalScheduled.get();
}
- private void removeMessages(List<Long> messageIds)
+ void removeMessages(List<Long> messageIds)
{
if(messageIds != null && !messageIds.isEmpty())
{
- String inpart = messageIds.stream().map(Object::toString).collect(Collectors.joining(", ", "(", ")"));
try(Connection conn = newConnection())
{
try
{
- try(Statement stmt = conn.createStatement())
- {
- int results = stmt.executeUpdate("DELETE FROM " + getMetaDataTableName()
- + " WHERE message_id IN " + inpart);
- stmt.close();
-
- if (results != messageIds.size())
- {
- getLogger().debug(
- "Some message ids in {} not found (attempt to remove failed - probably application initiated rollback)",
-
- messageIds);
- }
+ for (int i = 0; i <= messageIds.size() / IN_CLAUSE_MAX_SIZE; i++) {
+ List<Long> boundMessageIds = messageIds.stream()
+ .skip(i * IN_CLAUSE_MAX_SIZE)
+ .limit(IN_CLAUSE_MAX_SIZE)
+ .collect(Collectors.toList());
- getLogger().debug("Deleted metadata for messages {}", messageIds);
+ removeMessagesFromDatabase(conn, boundMessageIds);
}
-
- try(Statement stmt = conn.createStatement())
- {
- int results = stmt.executeUpdate("DELETE FROM " + getMessageContentTableName()
- + " WHERE message_id IN " + inpart);
- }
- conn.commit();
}
catch(SQLException e)
{
@@ -510,7 +496,31 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
throw new StoreException("Error removing messages with ids " + messageIds + " from database: " + e.getMessage(), e);
}
}
+ }
+ void removeMessagesFromDatabase(Connection conn, List<Long> messageIds) throws SQLException {
+ String inpart = messageIds.stream().map(Object::toString).collect(Collectors.joining(", ", "(", ")"));
+
+ try(Statement stmt = conn.createStatement())
+ {
+ int results = stmt.executeUpdate("DELETE FROM " + getMetaDataTableName() + " WHERE message_id IN " + inpart);
+ stmt.close();
+
+ if (results != messageIds.size())
+ {
+ getLogger().debug(
+ "Some message ids in {} not found (attempt to remove failed - probably application initiated rollback)",
+ messageIds);
+ }
+ getLogger().debug("Deleted metadata for messages {}", messageIds);
+ }
+
+ try(Statement stmt = conn.createStatement())
+ {
+ stmt.executeUpdate("DELETE FROM " + getMessageContentTableName() + " WHERE message_id IN " + inpart);
+ getLogger().debug("Deleted content for messages {}", messageIds);
+ }
+ conn.commit();
}
/**
diff --git a/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreTest.java b/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreTest.java
index 8be3376..84b93a2 100644
--- a/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreTest.java
+++ b/broker-plugins/jdbc-store/src/test/java/org/apache/qpid/server/store/jdbc/JDBCMessageStoreTest.java
@@ -25,15 +25,22 @@ import static org.apache.qpid.server.store.jdbc.TestJdbcUtils.getTableNames;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.Collection;
+import java.util.Collections;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import java.util.stream.LongStream;
import com.google.common.util.concurrent.ListenableFuture;
import org.junit.After;
@@ -51,6 +58,7 @@ import org.apache.qpid.server.store.MessageStoreTestCase;
import org.apache.qpid.server.store.Transaction;
import org.apache.qpid.server.store.TransactionLogResource;
import org.apache.qpid.server.virtualhost.jdbc.JDBCVirtualHost;
+import org.mockito.Mockito;
public class JDBCMessageStoreTest extends MessageStoreTestCase
{
@@ -169,6 +177,19 @@ public class JDBCMessageStoreTest extends MessageStoreTestCase
assertEquals("Delete action was not invoked", true, deleted.get());
}
+ @Test
+ public void testRemoveMessages() throws Exception
+ {
+ GenericJDBCMessageStore store = spy((GenericJDBCMessageStore) getStore());
+ when(store.newConnection()).thenReturn(mock(Connection.class, Mockito.RETURNS_MOCKS));
+
+ store.removeMessages(LongStream.rangeClosed(1,2001).boxed().collect(Collectors.toList()));
+
+ verify(store).removeMessagesFromDatabase(any(Connection.class), eq(LongStream.rangeClosed(1,1000).boxed().collect(Collectors.toList())));
+ verify(store).removeMessagesFromDatabase(any(Connection.class), eq(LongStream.rangeClosed(1001,2000).boxed().collect(Collectors.toList())));
+ verify(store).removeMessagesFromDatabase(any(Connection.class), eq(Collections.singletonList(2001L)));
+ }
+
private InternalMessage addTestMessage(final MessageStore store,
final String transactionalLogName,
final String messageContent)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org