+ d.stop();
+ }
+ }
+
+ @Scheduled(fixedRate = CONSUMER_SUPERVISION_INTERVAL_MS)
+ public synchronized void restartNonRunningTasks() {
+ for (KafkaJobDataConsumer consumer : activeSubscriptions.values()) {
+ if (!consumer.isRunning()) {
+ consumer.start();
+ }