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/12/01 20:36:48 UTC
[qpid-broker-j] branch 7.1.x updated (f72353c -> 3928068)
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 f72353c Revert "QPID-8381: [Broker-J] Update reference file"
new 95a8379 QPID-8387: [Broker-J] Handle exceptions thrown on asynchromous message removal in JDBC-based message stores
new 3928068 QPID-8387: [Broker-J] Await for asyncroronous message removal tasks to complete on message store close
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 | 96 +++++++++++++++++++---
1 file changed, 83 insertions(+), 13 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-8387: [Broker-J] Await for
asyncroronous message removal tasks to complete on message store close
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 39280687e8149805da529a0920542ff68617c391
Author: Alex Rudyy <or...@apache.org>
AuthorDate: Sun Dec 1 18:41:22 2019 +0000
QPID-8387: [Broker-J] Await for asyncroronous message removal tasks to complete on message store close
(cherry picked from commit 7193e26f6fad83f5ce81fd9ef855edb0fd5594ea)
---
.../server/store/jdbc/AbstractJDBCMessageStore.java | 21 +++++++++++++++++++++
1 file changed, 21 insertions(+)
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 e1181b2..21c7d87 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
@@ -41,6 +41,7 @@ import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -91,6 +92,8 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
private static final int EXECUTOR_THREADS_DEFAULT = Runtime.getRuntime().availableProcessors();
private static final String EXECUTOR_THREADS = "qpid.jdbcstore.executorThreads";
+ private static final String EXECUTOR_SHUTDOWN_TIMEOUT = "qpid.jdbcstore.executorShutdownTimeoutInSeconds";
+ private static final int EXECUTOR_SHUTDOWN_TIMEOUT_DEFAULT = 5;
private static final int DB_VERSION = 8;
@@ -116,6 +119,7 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
protected abstract void checkMessageStoreOpen();
private ScheduledThreadPoolExecutor _executor;
private volatile int _inClauseMaxSize;
+ private volatile int _executorShutdownTimeOut;
public AbstractJDBCMessageStore()
{
@@ -230,6 +234,8 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
int corePoolSize = getContextValue(Integer.class, EXECUTOR_THREADS, EXECUTOR_THREADS_DEFAULT);
+ _executorShutdownTimeOut = getContextValue(Integer.class, EXECUTOR_SHUTDOWN_TIMEOUT, EXECUTOR_SHUTDOWN_TIMEOUT_DEFAULT);
+
_executor = new ScheduledThreadPoolExecutor(corePoolSize, new ThreadFactory()
{
private final AtomicInteger _count = new AtomicInteger();
@@ -259,6 +265,21 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
if(_executor != null)
{
_executor.shutdown();
+ if (_executorShutdownTimeOut > 0)
+ {
+ try
+ {
+ if (!_executor.awaitTermination(_executorShutdownTimeOut, TimeUnit.SECONDS))
+ {
+ _executor.shutdownNow();
+ }
+ }
+ catch (InterruptedException e)
+ {
+ getLogger().warn("Interrupted during store executor shutdown:", e);
+ Thread.currentThread().interrupt();
+ }
+ }
}
}
---------------------------------------------------------------------
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-8387: [Broker-J] Handle exceptions
thrown on asynchromous message removal in JDBC-based message stores
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 95a8379c41417a5de3fb4bc828bb0efda06491d3
Author: Alex Rudyy <or...@apache.org>
AuthorDate: Sun Dec 1 18:00:00 2019 +0000
QPID-8387: [Broker-J] Handle exceptions thrown on asynchromous message removal in JDBC-based message stores
(cherry picked from commit a597b9051f67287ab64cd7c1a966bfeab239088d)
---
.../store/jdbc/AbstractJDBCMessageStore.java | 75 ++++++++++++++++++----
1 file changed, 62 insertions(+), 13 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 df02e55..e1181b2 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
@@ -38,6 +38,7 @@ import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -108,6 +109,7 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
private final Set<StoredJDBCMessage<?>> _messages = Collections.newSetFromMap(new ConcurrentHashMap<>());
private final Set<MessageDeleteListener> _messageDeleteListeners = Collections.newSetFromMap(new ConcurrentHashMap<>());
private final Set<Action<Connection>> _deleteActions = Collections.newSetFromMap(new ConcurrentHashMap<>());
+ private final Thread.UncaughtExceptionHandler _uncaughtExceptionHandler;
protected abstract boolean isMessageStoreOpen();
@@ -117,6 +119,7 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
public AbstractJDBCMessageStore()
{
+ _uncaughtExceptionHandler = Thread.getDefaultUncaughtExceptionHandler();
}
protected void setMaximumMessageId()
@@ -453,21 +456,67 @@ public abstract class AbstractJDBCMessageStore implements MessageStore
{
if(_messageRemovalScheduled.compareAndSet(false, true))
{
- _executor.submit(() -> {
- List<Long> messageIds;
- do
- {
- messageIds = _messagesToDelete.getAndSet(EMPTY_LIST);
- removeMessages(messageIds);
- } while(!messageIds.isEmpty());
-
+ try
+ {
+ _executor.submit(this::removeScheduledMessages);
+ }
+ catch (RejectedExecutionException e)
+ {
_messageRemovalScheduled.set(false);
- if(!_messagesToDelete.get().isEmpty())
- {
- scheduleMessageRemoval();
- }
+ throw new IllegalStateException("Cannot schedule removal of messages", e);
+ }
+ }
+ }
- });
+ private void removeScheduledMessages()
+ {
+ try
+ {
+ removeScheduledMessagesAndRescheduleIfRequired();
+ }
+ catch (RuntimeException e)
+ {
+ handleExceptionOnScheduledMessageRemoval(e);
+ }
+ }
+
+ private void removeScheduledMessagesAndRescheduleIfRequired()
+ {
+ List<Long> messageIds;
+ try
+ {
+ do
+ {
+ messageIds = _messagesToDelete.getAndSet(EMPTY_LIST);
+ removeMessages(messageIds);
+ } while (!messageIds.isEmpty());
+ }
+ finally
+ {
+ _messageRemovalScheduled.set(false);
+ }
+ if (!_messagesToDelete.get().isEmpty() && isMessageStoreOpen())
+ {
+ scheduleMessageRemoval();
+ }
+ }
+
+ private void handleExceptionOnScheduledMessageRemoval(final RuntimeException e)
+ {
+ if (isMessageStoreOpen())
+ {
+ if (_uncaughtExceptionHandler == null)
+ {
+ getLogger().error("Unexpected exception on asynchronous message removal", e);
+ }
+ else
+ {
+ _uncaughtExceptionHandler.uncaughtException(Thread.currentThread(), e);
+ }
+ }
+ else
+ {
+ getLogger().warn("Ignoring unexpected exception on asynchronous message removal as store is not open", e);
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org