diff --git a/src/main/java/org/apache/sling/event/impl/jobs/queues/JobQueueImpl.java b/src/main/java/org/apache/sling/event/impl/jobs/queues/JobQueueImpl.java
index be6f485..da1742c 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/queues/JobQueueImpl.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/queues/JobQueueImpl.java
@@ -107,6 +107,12 @@
     /** Semaphore for handling the max number of jobs. */
     private final Semaphore available;
 
+    /** original value of maxParallel with which the queue was created */
+    private final int maxParallel;
+
+    /** Semaphore for handling reduced number of available slots that are yet to drain */
+    private final Semaphore drainage;
+
     /** Guard for having only one thread executing start jobs. */
     private final AtomicBoolean startJobsGuard = new AtomicBoolean(false);
 
@@ -123,6 +129,7 @@
      * @param config The queue configuration
      * @param services The queue services
      * @param topics The topics handled by this queue
+     * @param outdatedQueueInfo
      * @param haltedTopics reference to pass newly halted topics back
      *
      * @return {@code JobQueueImpl} if there are jobs to process, {@code null} otherwise.
@@ -131,7 +138,8 @@
                         final InternalQueueConfiguration config,
                         final QueueServices services,
                         final Set<String> topics,
-                        final Set<String> haltedTopicsBackRef) {
+                        final Set<String> haltedTopicsBackRef,
+                        final OutdatedJobQueueInfo outdatedQueueInfo) {
         final QueueJobCache cache = new QueueJobCache(services.configuration, name, services.statisticsManager, config.getType(), topics);
         // the cache might contain newly halted topics.
         // these we need to retain, to avoid scanning them going forward.
@@ -149,7 +157,7 @@
         if ( cache.isEmpty() ) {
             return null;
         }
-        return new JobQueueImpl(name, config, services, cache);
+        return new JobQueueImpl(name, config, services, cache, outdatedQueueInfo);
     }
 
     /**
@@ -159,11 +167,13 @@
      * @param config The queue configuration
      * @param services The queue services
      * @param cache The job cache
+     * @param outdatedQueue
      */
     private JobQueueImpl(final String name,
                         final InternalQueueConfiguration config,
                         final QueueServices services,
-                        final QueueJobCache cache) {
+                        final QueueJobCache cache,
+                        final OutdatedJobQueueInfo outdatedQueue) {
         if ( config.getOwnThreadPoolSize() > 0 ) {
             this.threadPool = new EventingThreadPool(services.threadPoolManager, config.getOwnThreadPoolSize());
         } else {
@@ -175,7 +185,40 @@
         this.logger = LoggerFactory.getLogger(this.getClass().getName() + '.' + name);
         this.running = true;
         this.cache = cache;
-        this.available = new Semaphore(config.getMaxParallel(), true);
+        this.maxParallel = config.getMaxParallel();
+        if (outdatedQueue == null) {
+            // queue is created the first time
+            this.available = new Semaphore(this.maxParallel, true);
+            this.drainage = new Semaphore(0, true);
+        } else {
+            // queue was previously outdated - let's reuse available and drainage
+            this.available = outdatedQueue.getAvailable();
+            this.drainage = outdatedQueue.getDrainage();
+            int oldMaxParallel = outdatedQueue.getMaxParallel();
+            int maxParallelDiff = this.maxParallel - oldMaxParallel;
+            int drainedOldDrainage = 0;
+            int drainedOldAvailable = 0;
+            if (maxParallelDiff != 0) {
+                // config change
+                drainedOldDrainage = this.drainage.drainPermits();
+                drainedOldAvailable = this.available.drainPermits();
+                int netNewPermits = drainedOldAvailable - drainedOldDrainage + maxParallelDiff;
+                if (netNewPermits > 0) {
+                    this.available.release(netNewPermits);
+                } else if (netNewPermits < 0) {
+                    // special case : maxparallel got reduced since last outdating,
+                    // resulting in effectively negative number of currently available permits.
+                    // to account for that, jobs try to drain first before re-adding to available
+                    // to trigger this behaviour, releasing the permit-diff to drainage
+                    this.drainage.release(-netNewPermits);
+                }
+            }
+            logger.info("<init> reused outdated queue info: queueName : {}"
+                    + ", old available : {}, old drainage : {}, old maxParallel : {}"
+                    + ", new available : {}, new drainage : {}, new maxParallel : {}",
+                    queueName, drainedOldAvailable, drainedOldDrainage, oldMaxParallel,
+                    available.availablePermits(), drainage.availablePermits(), this.maxParallel);
+        }
         logger.info("Starting job queue {}", queueName);
         logger.debug("Configuration for job queue={}", configuration);
     }
@@ -345,7 +388,18 @@
                 this.logger.error("Exception during job processing.", re);
             }
         } finally {
-            this.available.release();
+            // try draining first
+            if (this.drainage.tryAcquire()) {
+                // special case : if drainage is used, this means maxparallel
+                // got reconfigured and we are not releasing a permit to
+                // available here, but instead reduce drainage.
+                final int approxPermits = this.drainage.availablePermits();
+                this.logger.debug("startJobHandler: drained 1 permit for {}, approx left to drain: {}",
+                        queueName, approxPermits);
+            } else {
+                // otherwise release as usual
+                this.available.release();
+            }
         }
     }
 
@@ -731,5 +785,9 @@
     QueueJobCache getCache() {
         return cache;
     }
+
+    OutdatedJobQueueInfo getOutdatedJobQueueInfo() {
+        return new OutdatedJobQueueInfo(available, maxParallel, drainage);
+    }
 }
 
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/queues/OutdatedJobQueueInfo.java b/src/main/java/org/apache/sling/event/impl/jobs/queues/OutdatedJobQueueInfo.java
new file mode 100644
index 0000000..dd6d4fa
--- /dev/null
+++ b/src/main/java/org/apache/sling/event/impl/jobs/queues/OutdatedJobQueueInfo.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sling.event.impl.jobs.queues;
+
+import java.util.concurrent.Semaphore;
+
+/**
+ * Encapsulates data required to survive queue outdating
+ */
+class OutdatedJobQueueInfo {
+
+    private final Semaphore available;
+    private final int maxParallel;
+    private final Semaphore drainage;
+
+    OutdatedJobQueueInfo(Semaphore available, int maxParallel, Semaphore drainage) {
+        this.available = available;
+        this.maxParallel = maxParallel;
+        this.drainage = drainage;
+    }
+
+    final Semaphore getAvailable() {
+        return available;
+    }
+
+    final int getMaxParallel() {
+        return maxParallel;
+    }
+
+    final Semaphore getDrainage() {
+        return drainage;
+    }
+}
diff --git a/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java b/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java
index 4336ad5..4485463 100644
--- a/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java
+++ b/src/main/java/org/apache/sling/event/impl/jobs/queues/QueueManager.java
@@ -128,6 +128,10 @@
     /** All active queues. */
     private final Map<String, JobQueueImpl> queues = new ConcurrentHashMap<>();
 
+    /** upon outdating a queue its available etc semaphores are parked here,
+     * to handle long-running jobs vs topology changes properly */
+    private final Map<String, OutdatedJobQueueInfo> outdatedQueues = new ConcurrentHashMap<>();
+
     /** We count the scheduler runs. */
     private volatile long schedulerRuns;
 
@@ -253,7 +257,8 @@
                 queue = null;
             }
             if ( queue == null ) {
-                queue = JobQueueImpl.createQueue(queueInfo.queueName, config, queueServices, filteredTopics, haltedTopics);
+                final OutdatedJobQueueInfo outdatedQueueInfo = outdatedQueues.get(queueInfo.queueName);
+                queue = JobQueueImpl.createQueue(queueInfo.queueName, config, queueServices, filteredTopics, haltedTopics, outdatedQueueInfo);
                 // on startup the queue might be empty and we get null back from createQueue
                 if ( queue != null ) {
                     isNewQueue = true;
@@ -285,7 +290,10 @@
         // remove the queue with the old name
         // check for main queue
         final String oldName = ResourceHelper.filterQueueName(queue.getName());
-        this.queues.remove(oldName);
+        final JobQueueImpl oldQueue = this.queues.remove(oldName);
+        if (oldQueue != null) {
+            outdatedQueues.put(oldName, oldQueue.getOutdatedJobQueueInfo());
+        }
         // check if we can close or have to rename
         if ( queue.tryToClose() ) {
             // copy statistics
diff --git a/src/test/java/org/apache/sling/event/impl/discovery/InitDelayingTopologyEventListenerTest.java b/src/test/java/org/apache/sling/event/impl/discovery/InitDelayingTopologyEventListenerTest.java
index 83a3ac4..b8f4772 100644
--- a/src/test/java/org/apache/sling/event/impl/discovery/InitDelayingTopologyEventListenerTest.java
+++ b/src/test/java/org/apache/sling/event/impl/discovery/InitDelayingTopologyEventListenerTest.java
@@ -299,7 +299,7 @@
     @Test
     public void testProperties() throws Exception {
         final TestListener delegate = new TestListener();
-        InitDelayingTopologyEventListener listener = new InitDelayingTopologyEventListener(1, delegate);
+        InitDelayingTopologyEventListener listener = new InitDelayingTopologyEventListener(2, delegate);
         listener.handleTopologyEvent(createEvent(Type.TOPOLOGY_INIT));
         listener.handleTopologyEvent(createEvent(Type.TOPOLOGY_CHANGING));
         listener.handleTopologyEvent(createEvent(Type.TOPOLOGY_CHANGED));
diff --git a/src/test/java/org/apache/sling/event/it/AbstractJobHandlingTest.java b/src/test/java/org/apache/sling/event/it/AbstractJobHandlingTest.java
index e8bc0ac..47931c6 100644
--- a/src/test/java/org/apache/sling/event/it/AbstractJobHandlingTest.java
+++ b/src/test/java/org/apache/sling/event/it/AbstractJobHandlingTest.java
@@ -112,7 +112,7 @@
 
         return options(
                 newConfiguration("org.apache.sling.event.impl.jobs.jcr.PersistenceHandler")
-                    .put(JobManagerConfiguration.PROPERTY_BACKGROUND_LOAD_DELAY, 3L)
+                    .put(JobManagerConfiguration.PROPERTY_BACKGROUND_LOAD_DELAY, backgroundLoadDelay())
                     .put("startup.delay", 1L)
                     .asOption(),
                 baseConfiguration(),
@@ -145,6 +145,10 @@
            );
     }
 
+    long backgroundLoadDelay() {
+        return 3L;
+    }
+
     protected ModifiableCompositeOption baseConfiguration() {
         return composite(
             failOnUnresolvedBundles(),
diff --git a/src/test/java/org/apache/sling/event/it/AbstractMaxParallelTest.java b/src/test/java/org/apache/sling/event/it/AbstractMaxParallelTest.java
new file mode 100644
index 0000000..cf3ec2f
--- /dev/null
+++ b/src/test/java/org/apache/sling/event/it/AbstractMaxParallelTest.java
@@ -0,0 +1,311 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sling.event.it;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotNull;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.sling.discovery.TopologyEvent;
+import org.apache.sling.discovery.TopologyEvent.Type;
+import org.apache.sling.discovery.TopologyEventListener;
+import org.apache.sling.discovery.TopologyView;
+import org.apache.sling.event.jobs.Job;
+import org.apache.sling.event.jobs.JobManager;
+import org.apache.sling.event.jobs.NotificationConstants;
+import org.apache.sling.event.jobs.consumer.JobConsumer;
+import org.junit.After;
+import org.osgi.framework.InvalidSyntaxException;
+import org.osgi.framework.ServiceReference;
+import org.osgi.framework.ServiceRegistration;
+import org.osgi.service.event.Event;
+import org.osgi.service.event.EventHandler;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public abstract class AbstractMaxParallelTest extends AbstractJobHandlingTest {
+
+    private final Logger log = LoggerFactory.getLogger(this.getClass());
+
+    private static final int BACKGROUND_LOAD_DELAY_SECONDS = 1;
+
+    private static final int EXTRA_CHAOS_DURATION_SECONDS = 20;
+
+    private static final int UNKNOWN_TOPOLOGY_FACTOR_MILLIS = 15;//100;
+
+    private static final int STABLE_TOPOLOGY_FACTOR_MILLIS = 40;//300;
+
+    static final String TOPIC_PREFIX = "sling/maxparallel/";
+
+    static final String TOPIC_NAME = TOPIC_PREFIX + "zero";
+
+    private final Object syncObj = new Object();
+
+    protected int max = -1;
+
+    @Override
+    long backgroundLoadDelay() {
+        return BACKGROUND_LOAD_DELAY_SECONDS;
+    }
+
+    @Override
+    @After
+    public void cleanup() {
+        super.cleanup();
+    }
+
+    private void registerMax(int cnt) {
+        synchronized(syncObj) {
+            max = Math.max(max, cnt);
+        }
+    }
+
+    /**
+     * Setup consumers
+     */
+    private void setupJobConsumers(long jobRunMillis) {
+        this.registerJobConsumer(TOPIC_NAME,
+
+            new JobConsumer() {
+
+                private AtomicInteger cnt = new AtomicInteger(0);
+
+                @Override
+                public JobResult process(final Job job) {
+                    int c = cnt.incrementAndGet();
+                    registerMax(c);
+                    log.info("process : start delaying. count=" + c + ", id="+ job.getId());
+                    try {
+                        Thread.sleep(jobRunMillis);
+                    } catch (InterruptedException e) {
+                        // TODO Auto-generated catch block
+                        e.printStackTrace();
+                    }
+                    log.info("process : done delaying. count=" + c + ", id="+ job.getId());
+                    cnt.decrementAndGet();
+                    return JobResult.OK;
+                }
+            });
+    }
+
+    private static final class CreateJobThread extends Thread {
+
+        private final Logger log = LoggerFactory.getLogger(this.getClass());
+
+        private final JobManager jobManager;
+
+        final AtomicLong finishedThreads;
+
+        private final Map<String, AtomicLong> created;
+
+        private final int numJobs;
+
+        public CreateJobThread(final JobManager jobManager,
+                Map<String, AtomicLong> created,
+                final AtomicLong finishedThreads,
+                int numJobs) {
+            this.jobManager = jobManager;
+            this.created = created;
+            this.finishedThreads = finishedThreads;
+            this.numJobs = numJobs;
+        }
+
+        @Override
+        public void run() {
+            AtomicInteger cnt = new AtomicInteger(0);
+            for(int i=0; i<numJobs; i++) {
+                final int c = cnt.incrementAndGet();
+                log.info("run: creating job " + c + " on topic " + TOPIC_NAME);
+                if (jobManager.addJob(TOPIC_NAME, null) != null) {
+                    created.get(TOPIC_NAME).incrementAndGet();
+                }
+            }
+            finishedThreads.incrementAndGet();
+        }
+
+    }
+
+    /**
+     * Setup chaos thread(s)
+     *
+     * Chaos is right now created by sending topology changing/changed events randomly
+     */
+    private void setupChaosThreads(final List<Thread> threads,
+            final AtomicLong finishedThreads, long duration) {
+        final List<TopologyView> views = new ArrayList<>();
+        // register topology listener
+        final ServiceRegistration<TopologyEventListener> reg = this.bc.registerService(TopologyEventListener.class, new TopologyEventListener() {
+
+            @Override
+            public void handleTopologyEvent(final TopologyEvent event) {
+                if ( event.getType() == Type.TOPOLOGY_INIT ) {
+                    views.add(event.getNewView());
+                }
+            }
+        }, null);
+        while ( views.isEmpty() ) {
+            this.sleep(10);
+        }
+        reg.unregister();
+        final TopologyView view = views.get(0);
+
+        try {
+            final Collection<ServiceReference<TopologyEventListener>> refs = this.bc.getServiceReferences(TopologyEventListener.class, null);
+            assertNotNull(refs);
+            assertFalse(refs.isEmpty());
+            TopologyEventListener found = null;
+            for(final ServiceReference<TopologyEventListener> ref : refs) {
+                final TopologyEventListener listener = this.bc.getService(ref);
+                if ( listener != null && listener.getClass().getName().equals("org.apache.sling.event.impl.jobs.config.TopologyHandler") ) {
+                    found = listener;
+                    break;
+                }
+                bc.ungetService(ref);
+            }
+            assertNotNull(found);
+            final TopologyEventListener tel = found;
+            log.info("setupChaosThreads : simulating TOPOLOGY_INIT");
+            tel.handleTopologyEvent(new TopologyEvent(Type.TOPOLOGY_INIT, null, view));
+
+            threads.add(new Thread() {
+
+                private final Random random = new Random();
+
+                @Override
+                public void run() {
+                    final long startTime = System.currentTimeMillis();
+                    // this thread runs 30 seconds longer than the job creation thread
+                    final long endTime = startTime + (duration +EXTRA_CHAOS_DURATION_SECONDS) * 1000;
+                    while ( System.currentTimeMillis() < endTime ) {
+                        final int sleepTime = random.nextInt(25) + 15;
+                        try {
+                            Thread.sleep(sleepTime * STABLE_TOPOLOGY_FACTOR_MILLIS);
+                        } catch ( final InterruptedException ie) {
+                            Thread.currentThread().interrupt();
+                        }
+                        log.info("setupChaosThreads : simulating TOPOLOGY_CHANGING");
+                        tel.handleTopologyEvent(new TopologyEvent(Type.TOPOLOGY_CHANGING, view, null));
+                        final int changingTime = random.nextInt(20) + 3;
+                        try {
+                            Thread.sleep(changingTime * UNKNOWN_TOPOLOGY_FACTOR_MILLIS);
+                        } catch ( final InterruptedException ie) {
+                            Thread.currentThread().interrupt();
+                        }
+                        log.info("setupChaosThreads : simulating TOPOLOGY_CHANGED");
+                        tel.handleTopologyEvent(new TopologyEvent(Type.TOPOLOGY_CHANGED, view, view));
+                    }
+                    tel.getClass().getName();
+                    finishedThreads.incrementAndGet();
+                }
+            });
+        } catch (InvalidSyntaxException e) {
+            e.printStackTrace();
+        }
+    }
+
+    void doTestMaxParallel(int numJobs, long jobRunMillis, long duration) throws Exception {
+        final JobManager jobManager = this.getJobManager();
+
+        final Map<String, AtomicLong> added = new HashMap<>();
+        final Map<String, AtomicLong> created = new HashMap<>();
+        final Map<String, AtomicLong> finished = new HashMap<>();
+        final List<String> topics = new ArrayList<>();
+        added.put(TOPIC_NAME, new AtomicLong());
+        created.put(TOPIC_NAME, new AtomicLong());
+        finished.put(TOPIC_NAME, new AtomicLong());
+        topics.add(TOPIC_NAME);
+
+        final List<Thread> threads = new ArrayList<>();
+        final AtomicLong finishedThreads = new AtomicLong();
+
+        this.registerEventHandler("org/apache/sling/event/notification/job/*",
+                new EventHandler() {
+
+                    @Override
+                    public void handleEvent(final Event event) {
+                        final String topic = (String) event.getProperty(NotificationConstants.NOTIFICATION_PROPERTY_JOB_TOPIC);
+                        if ( NotificationConstants.TOPIC_JOB_FINISHED.equals(event.getTopic())) {
+                            finished.get(topic).incrementAndGet();
+                        } else if ( NotificationConstants.TOPIC_JOB_ADDED.equals(event.getTopic())) {
+                            added.get(topic).incrementAndGet();
+                        }
+                    }
+                });
+
+        // setup job consumers
+        this.setupJobConsumers(jobRunMillis);
+
+        // setup job creation tests
+        new CreateJobThread(jobManager, created, finishedThreads, numJobs).start();
+
+        // wait until 1 job is being processed
+        log.info("doTestMaxParallel : waiting until 1 job is being processed");
+        while ( max <= 0 ) {
+            this.sleep(100);
+        }
+        log.info("doTestMaxParallel : 1 job was processed, ready to go. max=" + max);
+
+        this.setupChaosThreads(threads, finishedThreads, duration);
+
+        log.info("doTestMaxParallel : starting threads (" + threads.size() + ")");
+        // start threads
+        for(final Thread t : threads) {
+            t.setDaemon(true);
+            t.start();
+        }
+
+        log.info("doTestMaxParallel: sleeping for " + duration + " seconds to wait for threads to finish...");
+        // for sure we can sleep for the duration
+        this.sleep(duration * 1000);
+
+        log.info("doTestMaxParallel: polling for threads to finish...");
+        // wait until threads are finished
+        while ( finishedThreads.get() < threads.size() ) {
+            this.sleep(100);
+        }
+
+        final Set<String> allTopics = new HashSet<>(topics);
+        log.info("doTestMaxParallel: waiting for job handling to finish... " + allTopics.size());
+        while ( !allTopics.isEmpty() ) {
+            final Iterator<String> iter = allTopics.iterator();
+            while ( iter.hasNext() ) {
+                final String topic = iter.next();
+                log.info("doTestMaxParallel: checking topic= " + topic +
+                        ", finished=" + finished.get(topic).get() + ", created=" + created.get(topic).get());
+                if ( finished.get(topic).get() == created.get(topic).get() ) {
+                    iter.remove();
+                }
+            }
+            log.info("doTestMaxParallel: waiting for job handling to finish... " + allTopics.size());
+            this.sleep(1000);
+        }
+        log.info("doTestMaxParallel: done.");
+    }
+}
diff --git a/src/test/java/org/apache/sling/event/it/OrderedMaxParallelTest.java b/src/test/java/org/apache/sling/event/it/OrderedMaxParallelTest.java
new file mode 100644
index 0000000..03085cd
--- /dev/null
+++ b/src/test/java/org/apache/sling/event/it/OrderedMaxParallelTest.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sling.event.it;
+
+import static org.junit.Assert.assertEquals;
+
+import java.io.IOException;
+import java.util.Dictionary;
+import java.util.Hashtable;
+
+import org.apache.sling.event.impl.jobs.config.ConfigurationConstants;
+import org.apache.sling.event.jobs.QueueConfiguration;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.junit.PaxExam;
+
+@RunWith(PaxExam.class)
+public class OrderedMaxParallelTest extends AbstractMaxParallelTest {
+
+    private static final int DURATION = 40;
+
+    @Override
+    @Before
+    public void setup() throws IOException {
+        super.setup();
+
+        // create ordered test queue
+        final org.osgi.service.cm.Configuration orderedConfig = this.configAdmin.createFactoryConfiguration("org.apache.sling.event.jobs.QueueConfiguration", null);
+        final Dictionary<String, Object> orderedProps = new Hashtable<>();
+        orderedProps.put(ConfigurationConstants.PROP_NAME, "ordered-max-parallel");
+        orderedProps.put(ConfigurationConstants.PROP_TYPE, QueueConfiguration.Type.ORDERED.name());
+        orderedProps.put(ConfigurationConstants.PROP_TOPICS, TOPIC_NAME);
+        orderedProps.put(ConfigurationConstants.PROP_RETRIES, 2);
+        orderedProps.put(ConfigurationConstants.PROP_RETRY_DELAY, 2000L);
+        orderedProps.put(ConfigurationConstants.PROP_MAX_PARALLEL, 1);
+        orderedConfig.update(orderedProps);
+
+        this.sleep(1000L);
+    }
+
+    @Test(timeout=DURATION * 16000L)
+    public void testOrderedMaxParallel_slow() throws Exception {
+        doTestMaxParallel(12, 1717, DURATION);
+
+        assertEquals(1, max);
+    }
+
+    @Test(timeout=DURATION * 16000L)
+    public void testOrderedMaxParallel2_fast() throws Exception {
+        doTestMaxParallel(50, 123, DURATION);
+
+        assertEquals(1, max);
+    }
+}
diff --git a/src/test/java/org/apache/sling/event/it/RoundRobinMaxParallelTest.java b/src/test/java/org/apache/sling/event/it/RoundRobinMaxParallelTest.java
new file mode 100644
index 0000000..7e5531b
--- /dev/null
+++ b/src/test/java/org/apache/sling/event/it/RoundRobinMaxParallelTest.java
@@ -0,0 +1,73 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.sling.event.it;
+
+import static org.junit.Assert.assertTrue;
+
+import java.io.IOException;
+import java.util.Dictionary;
+import java.util.Hashtable;
+
+import org.apache.sling.event.impl.jobs.config.ConfigurationConstants;
+import org.apache.sling.event.jobs.QueueConfiguration;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.ops4j.pax.exam.junit.PaxExam;
+
+@RunWith(PaxExam.class)
+public class RoundRobinMaxParallelTest extends AbstractMaxParallelTest {
+
+    private static final int MAX_PARALLEL = 3;
+
+    private static final int DURATION = 50;
+
+    @Override
+    @Before
+    public void setup() throws IOException {
+        super.setup();
+
+        // create ordered test queue
+        final org.osgi.service.cm.Configuration orderedConfig = this.configAdmin.createFactoryConfiguration("org.apache.sling.event.jobs.QueueConfiguration", null);
+        final Dictionary<String, Object> orderedProps = new Hashtable<>();
+        orderedProps.put(ConfigurationConstants.PROP_NAME, "round-robin-max-parallel");
+        orderedProps.put(ConfigurationConstants.PROP_TYPE, QueueConfiguration.Type.TOPIC_ROUND_ROBIN.name());
+        orderedProps.put(ConfigurationConstants.PROP_TOPICS, TOPIC_NAME);
+        orderedProps.put(ConfigurationConstants.PROP_RETRIES, 2);
+        orderedProps.put(ConfigurationConstants.PROP_RETRY_DELAY, 2000L);
+        orderedProps.put(ConfigurationConstants.PROP_MAX_PARALLEL, MAX_PARALLEL);
+        orderedConfig.update(orderedProps);
+
+        this.sleep(1000L);
+    }
+
+    @Test(timeout=DURATION * 16000L)
+    public void testRoundRobinMaxParallel_slow() throws Exception {
+        doTestMaxParallel(20, 1717, DURATION);
+
+        assertTrue(max <= MAX_PARALLEL);
+    }
+
+    @Test(timeout=DURATION * 16000L)
+    public void testRoundRobinMaxParallel_fast() throws Exception {
+        doTestMaxParallel(200, 123, DURATION);
+
+        assertTrue(max <= MAX_PARALLEL);
+    }
+}