- boolean stateChanged = running.compareAndSet(true, false);
- if (!stateChanged)
- throw new RuntimeException("Consumer instance " + id + " is not running!");
-
- log.info("{} - Stopping", id);
- consumer.wakeup();
- future.get();
- log.info("{} - Stopped - consumed {} messages so far", id, consumed);
+ lock.lock();
+ try
+ {
+ if (!running)
+ throw new IllegalStateException("Consumer instance " + id + " is not running!");
+
+ log.info("{} - Stopping", id);
+ consumer.wakeup();
+ condition.await();
+ log.info("{} - Stopped - consumed {} messages so far", id, consumed);
+ }
+ finally
+ {
+ lock.unlock();
+ }