You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by lq...@apache.org on 2017/10/10 09:19:24 UTC
qpid-broker-j git commit: QPID-7956: [Java Broker] Guard against the
possibility of a null housekeeper executer when reallocating messages
Repository: qpid-broker-j
Updated Branches:
refs/heads/master 1de6cd5dc -> 4ff194a7b
QPID-7956: [Java Broker] Guard against the possibility of a null housekeeper executer when reallocating messages
Project: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/commit/4ff194a7
Tree: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/tree/4ff194a7
Diff: http://git-wip-us.apache.org/repos/asf/qpid-broker-j/diff/4ff194a7
Branch: refs/heads/master
Commit: 4ff194a7b3e331bba1ea1e850a8d4a70859ad0e9
Parents: 1de6cd5
Author: Lorenz Quack <lq...@apache.org>
Authored: Tue Oct 10 10:18:42 2017 +0100
Committer: Lorenz Quack <lq...@apache.org>
Committed: Tue Oct 10 10:19:10 2017 +0100
----------------------------------------------------------------------
.../apache/qpid/server/queue/AbstractQueue.java | 22 --------
.../org/apache/qpid/server/txn/DtxRegistry.java | 1 +
.../server/virtualhost/AbstractVirtualHost.java | 55 +++++++++-----------
3 files changed, 26 insertions(+), 52 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/4ff194a7/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java b/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
index 91a1e2b..0c935fc 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/queue/AbstractQueue.java
@@ -27,7 +27,6 @@ import java.io.OutputStream;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.nio.charset.StandardCharsets;
-import java.security.AccessControlContext;
import java.security.AccessControlException;
import java.security.AccessController;
import java.security.Principal;
@@ -48,7 +47,6 @@ import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CopyOnWriteArrayList;
-import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.zip.GZIPInputStream;
@@ -663,26 +661,6 @@ public abstract class AbstractQueue<X extends AbstractQueue<X>>
// ------ Getters and Setters
- public void execute(final String name, Runnable runnable, AccessControlContext context)
- {
- try
- {
- if (_virtualHost.getState() != State.UNAVAILABLE)
- {
- _virtualHost.executeTask(name, runnable, context);
- }
- }
- catch (RejectedExecutionException ree)
- {
- // Ignore - QueueRunner submitted execution as queue was being stopped.
- if(!_stopped.get())
- {
- _logger.error("Unexpected rejected execution", ree);
- throw ree;
- }
- }
- }
-
@Override
public boolean isExclusive()
{
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/4ff194a7/broker-core/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java b/broker-core/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java
index 0fa4da2..3bcf108 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/txn/DtxRegistry.java
@@ -49,6 +49,7 @@ public class DtxRegistry
public ScheduledFuture<?> scheduleTask(final long delay, final Runnable task)
{
+ // TODO if the virtualhost is shutdown, the housekeeper may be null
return _virtualHost.scheduleTask(delay, task);
}
http://git-wip-us.apache.org/repos/asf/qpid-broker-j/blob/4ff194a7/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
----------------------------------------------------------------------
diff --git a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
index 33a8b4e..50c2f9c 100644
--- a/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
+++ b/broker-core/src/main/java/org/apache/qpid/server/virtualhost/AbstractVirtualHost.java
@@ -51,7 +51,6 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.UUID;
-import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
@@ -1216,15 +1215,6 @@ public abstract class AbstractVirtualHost<X extends AbstractVirtualHost<X>> exte
}
}
- protected void removeHouseKeepingTasks()
- {
- BlockingQueue<Runnable> taskQueue = _houseKeepingTaskExecutor.getQueue();
- for (final Runnable runnable : taskQueue)
- {
- _houseKeepingTaskExecutor.remove(runnable);
- }
- }
-
private void closeNetworkConnectionScheduler()
{
if(_networkConnectionScheduler != null)
@@ -1419,30 +1409,35 @@ public abstract class AbstractVirtualHost<X extends AbstractVirtualHost<X>> exte
@Override
public ListenableFuture<Void> reallocateMessages()
{
- final Future future;
- try
+ final ScheduledThreadPoolExecutor houseKeepingTaskExecutor = _houseKeepingTaskExecutor;
+ if (houseKeepingTaskExecutor != null)
{
- future = _houseKeepingTaskExecutor.submit(() ->
- {
- final Collection<Queue> queues = getChildren(Queue.class);
- for (Queue q : queues)
- {
- if (q.getState() == State.ACTIVE)
- {
- q.reallocateMessages();
- }
- }
- });
- return JdkFutureAdapters.listenInPoolThread(future);
- }
- catch (RejectedExecutionException e)
- {
- if (!_houseKeepingTaskExecutor.isShutdown())
+ try
{
- _logger.warn("Failed to schedule reallocation of messages", e);
+ final Future<Void> future = houseKeepingTaskExecutor.submit(() ->
+ {
+ final Collection<Queue> queues =
+ getChildren(Queue.class);
+ for (Queue q : queues)
+ {
+ if (q.getState() == State.ACTIVE)
+ {
+ q.reallocateMessages();
+ }
+ }
+ return null;
+ });
+ return JdkFutureAdapters.listenInPoolThread(future);
+ }
+ catch (RejectedExecutionException e)
+ {
+ if (!houseKeepingTaskExecutor.isShutdown())
+ {
+ _logger.warn("Failed to schedule reallocation of messages", e);
+ }
}
- return Futures.immediateFuture(null);
}
+ return Futures.immediateFuture(null);
}
@Override
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org