This is an automated email from the ASF dual-hosted git repository.
joerghoh pushed a commit to branch master
in repository
https://gitbox.apache.org/repos/asf/sling-org-apache-sling-event.git
The following commit(s) were added to refs/heads/master by this push:
new d66a8f0 SLING-13243 make topology-chaos ITs wait on completion
instead of fixed sleeps
d66a8f0 is described below
commit d66a8f08d4b48ba8523b9cfff3b5019e989f0ce1
Author: Jörg Hoh <[email protected]>
AuthorDate: Sun Jun 21 14:31:52 2026 +0200
SLING-13243 make topology-chaos ITs wait on completion instead of fixed
sleeps
RoundRobinMaxParallelIT (via AbstractMaxParallelIT) and ChaosIT blocked for
a hardcoded duration plus a fixed chaos tail regardless of actual progress,
making them slow and their runtime independent of the real workload.
Replace the blind sleeps with CountDownLatch-based waits so the tests
proceed
as soon as the work is done, and bound the remaining waits so a real hang
fails fast with a clear message instead of running to the @Test timeout.
Functional intent of the tests is unchanged.
- AbstractMaxParallelIT: wait on jobsCreated/jobsCompleted latches
(bounded);
chaos thread stops a short grace period after jobs finish instead of a
fixed
tail; make `max` volatile; drop dead code; parameterized logging.
- ChaosIT: replace fixed sleep/poll with creationLatch + allThreadsLatch;
chunked, interruptible chaos sleep so the grace period is honored; bound
the
drain loop with progress logging so slow draining is distinguishable from
a
stall.
---
.../sling/event/it/AbstractMaxParallelIT.java | 125 +++++++++--------
.../java/org/apache/sling/event/it/ChaosIT.java | 149 +++++++++++++++------
2 files changed, 171 insertions(+), 103 deletions(-)
diff --git a/src/test/java/org/apache/sling/event/it/AbstractMaxParallelIT.java
b/src/test/java/org/apache/sling/event/it/AbstractMaxParallelIT.java
index 0117267..d4c2e22 100644
--- a/src/test/java/org/apache/sling/event/it/AbstractMaxParallelIT.java
+++ b/src/test/java/org/apache/sling/event/it/AbstractMaxParallelIT.java
@@ -21,12 +21,11 @@ package org.apache.sling.event.it;
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.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
@@ -48,12 +47,14 @@ import org.slf4j.LoggerFactory;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
public abstract class AbstractMaxParallelIT extends AbstractJobHandlingIT {
private static final int BACKGROUND_LOAD_DELAY_SECONDS = 1;
- private static final int EXTRA_CHAOS_DURATION_SECONDS = 20;
+ /** Grace period (in seconds) the chaos thread keeps running after all
jobs have finished. */
+ private static final int CHAOS_GRACE_SECONDS = 5;
private static final int UNKNOWN_TOPOLOGY_FACTOR_MILLIS = 15; // 100;
@@ -65,7 +66,7 @@ public abstract class AbstractMaxParallelIT extends
AbstractJobHandlingIT {
private final Object syncObj = new Object();
- protected int max = -1;
+ protected volatile int max = -1;
@Override
protected long backgroundLoadDelay() {
@@ -81,24 +82,23 @@ public abstract class AbstractMaxParallelIT extends
AbstractJobHandlingIT {
/**
* Setup consumers
*/
- private void setupJobConsumers(long jobRunMillis) {
+ private void setupJobConsumers(long jobDuration) {
this.registerJobConsumer(TOPIC_NAME, new JobConsumer() {
- private AtomicInteger cnt = new AtomicInteger(0);
+ private AtomicInteger concurrentExecutionsCounter = new
AtomicInteger(0);
@Override
public JobResult process(final Job job) {
- int c = cnt.incrementAndGet();
+ int c = concurrentExecutionsCounter.incrementAndGet();
registerMax(c);
- log.info("process : start delaying. count=" + c + ", id=" +
job.getId());
+ log.info("process : start delaying. concurrentExecutions={},
id={}", c, job.getId());
try {
- Thread.sleep(jobRunMillis);
+ Thread.sleep(jobDuration);
} catch (InterruptedException e) {
- // TODO Auto-generated catch block
e.printStackTrace();
}
- log.info("process : done delaying. count=" + c + ", id=" +
job.getId());
- cnt.decrementAndGet();
+ log.info("process : done delaying. concurrentExecutions={},
id={}", c, job.getId());
+ concurrentExecutionsCounter.decrementAndGet();
return JobResult.OK;
}
});
@@ -110,20 +110,13 @@ public abstract class AbstractMaxParallelIT extends
AbstractJobHandlingIT {
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) {
+ public CreateJobThread(final JobManager jobManager, Map<String,
AtomicLong> created, int numJobs) {
this.jobManager = jobManager;
this.created = created;
- this.finishedThreads = finishedThreads;
this.numJobs = numJobs;
}
@@ -132,12 +125,11 @@ public abstract class AbstractMaxParallelIT extends
AbstractJobHandlingIT {
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);
+ log.info("run: creating job {} on topic {}", c, TOPIC_NAME);
if (jobManager.addJob(TOPIC_NAME, null) != null) {
created.get(TOPIC_NAME).incrementAndGet();
}
}
- finishedThreads.incrementAndGet();
}
}
@@ -146,7 +138,8 @@ public abstract class AbstractMaxParallelIT extends
AbstractJobHandlingIT {
*
* Chaos is right now created by sending topology changing/changed events
randomly
*/
- private void setupChaosThreads(final List<Thread> threads, final
AtomicLong finishedThreads, long duration) {
+ private void setupChaosThreads(
+ final List<Thread> threads, final CountDownLatch jobsLatch, final
CountDownLatch chaosLatch) {
final List<TopologyView> views = new ArrayList<>();
// register topology listener
final ServiceRegistration<TopologyEventListener> reg =
this.bundleContext.registerService(
@@ -189,16 +182,23 @@ public abstract class AbstractMaxParallelIT extends
AbstractJobHandlingIT {
log.info("setupChaosThreads : simulating TOPOLOGY_INIT");
tel.handleTopologyEvent(new TopologyEvent(Type.TOPOLOGY_INIT,
null, view));
- threads.add(new Thread() {
+ threads.add(new Thread("topology-changer") {
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) {
+ long graceDeadline = -1;
+ while (true) {
+ if (jobsLatch.getCount() == 0) {
+ // keep creating chaos while jobs are still being
processed and for a short
+ // grace period afterwards
+ if (graceDeadline < 0) {
+ graceDeadline = System.currentTimeMillis() +
CHAOS_GRACE_SECONDS * 1000L;
+ } else if (System.currentTimeMillis() >=
graceDeadline) {
+ break;
+ }
+ }
final int sleepTime = random.nextInt(25) + 15;
try {
Thread.sleep(sleepTime *
STABLE_TOPOLOGY_FACTOR_MILLIS);
@@ -216,8 +216,7 @@ public abstract class AbstractMaxParallelIT extends
AbstractJobHandlingIT {
log.info("setupChaosThreads : simulating
TOPOLOGY_CHANGED");
tel.handleTopologyEvent(new
TopologyEvent(Type.TOPOLOGY_CHANGED, view, view));
}
- tel.getClass().getName();
- finishedThreads.incrementAndGet();
+ chaosLatch.countDown();
}
});
} catch (InvalidSyntaxException e) {
@@ -230,14 +229,17 @@ public abstract class AbstractMaxParallelIT extends
AbstractJobHandlingIT {
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();
+ // count down for every job created
+ final CountDownLatch jobsCreatedLatch = new CountDownLatch(numJobs);
+ // counted down once for every finished job; lets us wait exactly
until all jobs are done
+ final CountDownLatch jobsCompletedLatch = new CountDownLatch(numJobs);
+ // counted down by the chaos thread once it has stopped
+ final CountDownLatch chaosLatch = new CountDownLatch(1);
this.registerEventHandler("org/apache/sling/event/notification/job/*",
new EventHandler() {
@@ -246,8 +248,10 @@ public abstract class AbstractMaxParallelIT extends
AbstractJobHandlingIT {
final String topic = (String)
event.getProperty(NotificationConstants.NOTIFICATION_PROPERTY_JOB_TOPIC);
if
(NotificationConstants.TOPIC_JOB_FINISHED.equals(event.getTopic())) {
finished.get(topic).incrementAndGet();
+ jobsCompletedLatch.countDown();
} else if
(NotificationConstants.TOPIC_JOB_ADDED.equals(event.getTopic())) {
added.get(topic).incrementAndGet();
+ jobsCreatedLatch.countDown();
}
}
});
@@ -256,50 +260,43 @@ public abstract class AbstractMaxParallelIT extends
AbstractJobHandlingIT {
this.setupJobConsumers(jobRunMillis);
// setup job creation tests
- new CreateJobThread(jobManager, created, finishedThreads,
numJobs).start();
+ new CreateJobThread(jobManager, created, numJobs).start();
// wait until 1 job is being processed
- log.info("doTestMaxParallel : waiting until 1 job is being processed");
+ log.info("doTestMaxParallel : waiting until at least 1 job is being
processed");
while (max <= 0) {
this.sleep(100);
}
- log.info("doTestMaxParallel : 1 job was processed, ready to go. max="
+ max);
+ log.info("doTestMaxParallel : job processing started, ready to go");
- this.setupChaosThreads(threads, finishedThreads, duration);
+ this.setupChaosThreads(threads, jobsCompletedLatch, chaosLatch);
- log.info("doTestMaxParallel : starting threads (" + threads.size() +
")");
+ 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);
- }
+ // a generous timeout to wait until a jobs are created
+ final long timeoutMillis = duration * 4 * 1000L;
+ log.info("doTestMaxParallel: waiting for all {} jobs to be
created...", numJobs);
+ assertTrue(
+ "Not all " + numJobs + " jobs created within " + (duration *
4) + " seconds",
+ jobsCreatedLatch.await(timeoutMillis, TimeUnit.MILLISECONDS));
+ log.info("All jobs were created");
+
+ // generous timeout to fail fast instead of hanging if jobs get stuck;
job processing under
+ // topology chaos is much slower than the ideal throughput, so allow a
wide margin while
+ // still staying well below the JUnit @Test timeout
+ log.info("doTestMaxParallel: waiting for all {} jobs to finish...",
numJobs);
+ assertTrue(
+ "Not all " + numJobs + " jobs finished within " + (duration *
4) + " seconds",
+ jobsCompletedLatch.await(timeoutMillis,
TimeUnit.MILLISECONDS));
+
+ log.info("doTestMaxParallel: waiting for chaos thread to stop...");
+ assertTrue("Chaos thread did not stop in time",
chaosLatch.await(CHAOS_GRACE_SECONDS * 2L, TimeUnit.SECONDS));
- 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/ChaosIT.java
b/src/test/java/org/apache/sling/event/it/ChaosIT.java
index 98d0c6e..532a714 100644
--- a/src/test/java/org/apache/sling/event/it/ChaosIT.java
+++ b/src/test/java/org/apache/sling/event/it/ChaosIT.java
@@ -27,6 +27,8 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.sling.discovery.TopologyEvent;
@@ -54,6 +56,7 @@ import org.osgi.service.event.EventHandler;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
+import static org.junit.Assert.assertTrue;
import static org.ops4j.pax.exam.CoreOptions.options;
import static
org.ops4j.pax.exam.cm.ConfigurationAdminOptions.factoryConfiguration;
@@ -64,6 +67,12 @@ public class ChaosIT extends AbstractJobHandlingIT {
/** Duration for firing jobs in seconds. */
private static final long DURATION = 1 * 60;
+ /** Grace period (in seconds) the chaos thread keeps running after job
creation has finished. */
+ private static final int CHAOS_GRACE_SECONDS = 5;
+
+ /** Maximum time (in seconds) to wait for all created jobs to finish
processing. */
+ private static final int JOB_DRAIN_TIMEOUT_SECONDS = 600;
+
private static final int NUM_ORDERED_THREADS = 3;
private static final int NUM_PARALLEL_THREADS = 6;
private static final int NUM_ROUND_THREADS = 6;
@@ -158,17 +167,21 @@ public class ChaosIT extends AbstractJobHandlingIT {
final Map<String, AtomicLong> created;
- final AtomicLong finishedThreads;
+ final CountDownLatch creationLatch;
+
+ final CountDownLatch allThreadsLatch;
public CreateJobThread(
final JobManager jobManager,
final String[] topics,
final Map<String, AtomicLong> created,
- final AtomicLong finishedThreads) {
+ final CountDownLatch creationLatch,
+ final CountDownLatch allThreadsLatch) {
this.topics = topics;
this.jobManager = jobManager;
this.created = created;
- this.finishedThreads = finishedThreads;
+ this.creationLatch = creationLatch;
+ this.allThreadsLatch = allThreadsLatch;
}
@Override
@@ -188,12 +201,13 @@ public class ChaosIT extends AbstractJobHandlingIT {
}
final int sleepTime = random.nextInt(200);
try {
- Thread.sleep(sleepTime);
- } catch (final InterruptedException ie) {
+ this.sleep(sleepTime);
+ } catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
- finishedThreads.incrementAndGet();
+ creationLatch.countDown();
+ allThreadsLatch.countDown();
}
}
@@ -204,15 +218,16 @@ public class ChaosIT extends AbstractJobHandlingIT {
final List<Thread> threads,
final JobManager jobManager,
final Map<String, AtomicLong> created,
- final AtomicLong finishedThreads) {
+ final CountDownLatch creationLatch,
+ final CountDownLatch allThreadsLatch) {
for (int i = 0; i < NUM_ORDERED_THREADS; i++) {
- threads.add(new CreateJobThread(jobManager, ORDERED_TOPICS,
created, finishedThreads));
+ threads.add(new CreateJobThread(jobManager, ORDERED_TOPICS,
created, creationLatch, allThreadsLatch));
}
for (int i = 0; i < NUM_PARALLEL_THREADS; i++) {
- threads.add(new CreateJobThread(jobManager, PARALLEL_TOPICS,
created, finishedThreads));
+ threads.add(new CreateJobThread(jobManager, PARALLEL_TOPICS,
created, creationLatch, allThreadsLatch));
}
for (int i = 0; i < NUM_ROUND_THREADS; i++) {
- threads.add(new CreateJobThread(jobManager, ROUND_TOPICS, created,
finishedThreads));
+ threads.add(new CreateJobThread(jobManager, ROUND_TOPICS, created,
creationLatch, allThreadsLatch));
}
}
@@ -221,7 +236,8 @@ public class ChaosIT extends AbstractJobHandlingIT {
*
* Chaos is right now created by sending topology changing/changed events
randomly
*/
- private void setupChaosThreads(final List<Thread> threads, final
AtomicLong finishedThreads) {
+ private void setupChaosThreads(
+ final List<Thread> threads, final CountDownLatch creationLatch,
final CountDownLatch allThreadsLatch) {
final List<TopologyView> views = new ArrayList<>();
// register topology listener
final ServiceRegistration<TopologyEventListener> reg =
this.bundleContext.registerService(
@@ -262,33 +278,56 @@ public class ChaosIT extends AbstractJobHandlingIT {
assertNotNull(found);
final TopologyEventListener tel = found;
- threads.add(new Thread() {
+ threads.add(new Thread("chaos-topology-change") {
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 + 30) * 1000;
- while (System.currentTimeMillis() < endTime) {
- final int sleepTime = random.nextInt(25) + 15;
- try {
- Thread.sleep(sleepTime * 1000);
- } catch (final InterruptedException ie) {
- Thread.currentThread().interrupt();
+ // keep creating chaos while jobs are being created and
for a short grace
+ // period afterwards, so the drain phase is exercised
under topology changes too
+ long graceDeadline = -1;
+ while (true) {
+ if (creationLatch.getCount() == 0) {
+ if (graceDeadline < 0) {
+ graceDeadline = System.currentTimeMillis() +
CHAOS_GRACE_SECONDS * 1000L;
+ } else if (System.currentTimeMillis() >=
graceDeadline) {
+ break;
+ }
}
+ final int sleepTime = random.nextInt(25) + 15;
+ chaosSleep(sleepTime * 1000L);
tel.handleTopologyEvent(new
TopologyEvent(Type.TOPOLOGY_CHANGING, view, null));
+ log.info("Sent TopologyEvent (newView = null)");
final int changingTime = random.nextInt(20) + 3;
+ chaosSleep(changingTime * 1000L);
+ tel.handleTopologyEvent(new
TopologyEvent(Type.TOPOLOGY_CHANGED, view, view));
+ log.info("Sent TopologyEvent (newView not null)");
+ }
+ allThreadsLatch.countDown();
+ }
+
+ /**
+ * Sleep up to {@code totalMillis}, but in small chunks so the
stop condition is
+ * re-checked frequently. Returns early once job creation has
finished, so the
+ * grace period is actually honoured instead of being masked
by a single long
+ * sleep. A thread interrupt also terminates the sleep instead
of busy-looping.
+ */
+ private void chaosSleep(final long totalMillis) {
+ long remaining = totalMillis;
+ while (remaining > 0) {
+ final long chunk = Math.min(remaining, 250L);
try {
- Thread.sleep(changingTime * 1000);
+ Thread.sleep(chunk);
} catch (final InterruptedException ie) {
Thread.currentThread().interrupt();
+ return;
+ }
+ remaining -= chunk;
+ if (creationLatch.getCount() == 0) {
+ return;
}
- tel.handleTopologyEvent(new
TopologyEvent(Type.TOPOLOGY_CHANGED, view, view));
}
- tel.getClass().getName();
- finishedThreads.incrementAndGet();
}
});
} catch (InvalidSyntaxException e) {
@@ -326,7 +365,14 @@ public class ChaosIT extends AbstractJobHandlingIT {
}
final List<Thread> threads = new ArrayList<>();
- final AtomicLong finishedThreads = new AtomicLong();
+ final int numCreationThreads = NUM_ORDERED_THREADS +
NUM_PARALLEL_THREADS + NUM_ROUND_THREADS;
+ // Signals that all job-creation threads have finished. This is the
chaos thread's only cue
+ // to wind down (arm its grace period and stop). It cannot key off
allThreadsLatch for this,
+ // since that latch includes the chaos thread itself and can never
reach zero from within it.
+ final CountDownLatch creationLatch = new
CountDownLatch(numCreationThreads);
+ // all creation threads plus the single chaos thread; awaited by the
test to know when every
+ // worker thread has terminated
+ final CountDownLatch allThreadsLatch = new
CountDownLatch(numCreationThreads + 1);
this.registerEventHandler("org/apache/sling/event/notification/job/*",
new EventHandler() {
@@ -345,29 +391,33 @@ public class ChaosIT extends AbstractJobHandlingIT {
this.setupJobConsumers();
// setup job creation tests
- this.setupJobCreationThreads(threads, jobManager, created,
finishedThreads);
+ this.setupJobCreationThreads(threads, jobManager, created,
creationLatch, allThreadsLatch);
- this.setupChaosThreads(threads, finishedThreads);
+ this.setupChaosThreads(threads, creationLatch, allThreadsLatch);
- System.out.println("Starting threads...");
+ log.info("Starting threads...");
// start threads
for (final Thread t : threads) {
t.setDaemon(true);
t.start();
}
- System.out.println("Sleeping for " + DURATION + " seconds to wait for
threads to finish...");
- // for sure we can sleep for the duration
- this.sleep(DURATION * 1000);
+ log.info("Waiting for threads to finish...");
+ // wait until all threads (job creation + chaos) have finished; the
creation threads run
+ // for DURATION seconds, the chaos thread for a short grace period
longer
+ final long threadTimeout = (DURATION + CHAOS_GRACE_SECONDS) * 1000L +
60_000L;
+ boolean allThreadsFinished = allThreadsLatch.await(threadTimeout,
TimeUnit.MILLISECONDS);
- System.out.println("Polling for threads to finish...");
- // wait until threads are finished
- while (finishedThreads.get() < threads.size()) {
- this.sleep(100);
- }
+ // there is a small race condition in here, .getCount() could be zero
already now
+ assertTrue(
+ "Job creation and chaos threads did not finish in time, still
waiting for " + allThreadsLatch.getCount()
+ + "threads",
+ allThreadsFinished);
- System.out.println("Waiting for job handling to finish...");
+ log.info("Waiting for job handling to finish...");
final Set<String> allTopics = new HashSet<>(topics);
+ final long drainDeadline = System.currentTimeMillis() +
JOB_DRAIN_TIMEOUT_SECONDS * 1000L;
+ long lastProgressLog = 0;
while (!allTopics.isEmpty()) {
final Iterator<String> iter = allTopics.iterator();
while (iter.hasNext()) {
@@ -376,8 +426,29 @@ public class ChaosIT extends AbstractJobHandlingIT {
iter.remove();
}
}
- this.sleep(100);
+ if (!allTopics.isEmpty()) {
+ final long now = System.currentTimeMillis();
+ // log progress every 5 seconds so a genuine stall can be told
apart from slow draining
+ if (now - lastProgressLog >= 5000) {
+ lastProgressLog = now;
+ long remaining = 0;
+ for (final String topic : allTopics) {
+ remaining +=
+ created.get(topic).get() -
finished.get(topic).get();
+ }
+ log.info(
+ "Still draining: {} topics pending, {} jobs
outstanding (created vs finished)",
+ allTopics.size(),
+ remaining);
+ }
+ assertTrue(
+ "Jobs did not finish within " +
JOB_DRAIN_TIMEOUT_SECONDS + " seconds; topics still pending: "
+ + allTopics,
+ now < drainDeadline);
+ this.sleep(100);
+ }
}
+ log.info("Completed");
/* We could try to enable this with Oak again - but right now JR
observation handler is too
* slow.
System.out.println("Checking notifications...");
